stacks: the box converts a PostgreSQL major as a guarded-update step (09 6.4 part 10, decisions 35/37/38)

A step whose ladder entry carries engine_conversion {service, engine, from, to}
converts the database: the old engine alone, the check (owners, roles,
extensions, per-table row counts), pg_dumpall validated by its completion line,
the volume emptied only after the undo copy's marker is validated again, the new
engine alone, the load with ON_ERROR_STOP, the check again + PG_VERSION. Any
failure goes to the existing undo; a restart during converting is undone.
A PostgreSQL major move without the mark is refused before anything moves.
The old datadir's copy is kept until a backup is proven after the conversion.
17 tests, 9 red-proofs (audits/night-2026-09-26/B/).

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

controller/internal/stacks/deploy.go

@@ -166,6 +166,9 @@ type AppConfig struct {
 	// LastAutoUpdate (v0.271.0, `09` §6.4 part 7) is the automatic leg's last step on this app — the
 	// line on the app page („Automatikus frissítés %s-kor — sikeres").
 	LastAutoUpdate *AutoUpdateRecord `yaml:"last_auto_update,omitempty" json:"last_auto_update,omitempty"`
+	// ConversionCopy (v0.273.0, `09` §6.4 part 10) is the OLD datadir's copy kept after a successful
+	// PostgreSQL major conversion, until a backup of the converted app is proven (ReleaseConversionCopies).
+	ConversionCopy *ConversionCopy `yaml:"conversion_copy,omitempty" json:"conversion_copy,omitempty"`
 }
 
 // InstalledImage is one compose service's observed image. See AppConfig.InstalledImages.

controller/internal/stacks/ladder.go

@@ -48,6 +48,10 @@ type LadderEntry struct {
 	// automatic leg never takes a step that needs a person, and takes a files-may-change step only when
 	// a fresh WHOLE copy exists. A person's press ignores both — the marks bind the leg only.
 	Marks LadderMarks `yaml:"marks" json:"marks"`
+	// EngineConversion (v0.273.0, `09` §6.4 part 10) is the harness's mark that this step converts a
+	// PostgreSQL major (pgconvert.go). omitempty: an entry without it prints exactly as before, so R-680's
+	// failed-step records (tied to LadderPrint) are not reset by this field's arrival.
+	EngineConversion *EngineConversion `yaml:"engine_conversion" json:"engine_conversion,omitempty"`
 }
 
 // LadderMarks is the `marks` object of a ladder entry. NeedsPerson is JSON null (nil) or the tester's
@@ -137,6 +141,9 @@ type LadderStep struct {
 	Meta string
 	// Why is one operator-English sentence for the log.
 	Why string
+	// Entry is the ladder entry this step applies (nil when the installed version matches no entry, or
+	// the template has no ladder) — its marks, including the engine conversion (v0.273.0).
+	Entry *LadderEntry
 }
 
 // nextLadderStep decides WHICH definition the next press pins, from the catalog template directory
@@ -172,7 +179,7 @@ func nextLadderStep(templateDir string, pinned map[string]string) (LadderStep, e
 	}
 	left := len(ladder) - idx
 	if idx == len(ladder)-1 {
-		return LadderStep{Index: idx, Left: left, Source: current, Meta: currentMeta,
+		return LadderStep{Index: idx, Left: left, Source: current, Meta: currentMeta, Entry: &ladder[idx],
 			Why: fmt.Sprintf("the last step (%d of %d) — the catalog's current definition", idx+1, len(ladder))}, nil
 	}
 	src := StepFile(templateDir, ladder[idx].To)
@@ -187,7 +194,7 @@ func nextLadderStep(templateDir string, pinned map[string]string) (LadderStep, e
 	if _, err := os.Stat(meta); err != nil {
 		meta = currentMeta // a step written before R-664: the template's own, said in the log
 	}
-	return LadderStep{Index: idx, Left: left, Source: src, Meta: meta,
+	return LadderStep{Index: idx, Left: left, Source: src, Meta: meta, Entry: &ladder[idx],
 		Why: fmt.Sprintf("step %d of %d: %s → %s, from %s (probe from %s)", idx+1, len(ladder), summarisePin(ladder[idx].From), summarisePin(ladder[idx].To), filepath.Base(src), filepath.Base(meta))}, nil
 }
 

controller/internal/stacks/manager.go

@@ -278,7 +278,11 @@ type Manager struct {
 	updateHealthFn func(ctx context.Context, name string, timeout time.Duration) (bool, string)
 	// v0.263.0 undo seams (undo.go): the volume copier (nil ⇒ docker) and the undo's health wait,
 	// which receives the probe it must use (nil ⇒ waitUpdateHealthyMeta).
-	undoCopier         volumeCopier
+	undoCopier volumeCopier
+	// v0.273.0 conversion seams (pgconvert.go): the PostgreSQL process boundary (nil ⇒ docker) and the free
+	// space beside the stack dir (nil ⇒ statfs).
+	pgConv             pgConverter
+	convertFreeFn      func(path string) (int64, bool)
 	updateUndoHealthFn func(ctx context.Context, name string, timeout time.Duration, meta *Metadata) (bool, string)
 	probeRunFn         func(t probeTarget) *HealthProbeResult // the health wait's network probe; nil ⇒ runChecks
 	updateEventSink    func(UpdateEvent)                      // v0.264.0: the notifier; nil ⇒ no events

controller/internal/stacks/pgconvert.go

@@ -0,0 +1,864 @@
+package stacks
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"regexp"
+	"sort"
+	"strconv"
+	"strings"
+	"syscall"
+	"time"
+
+	"gitea.dooplex.hu/admin/felhom-controller/internal/dockerexec"
+	"gopkg.in/yaml.v3"
+)
+
+// ── The PostgreSQL major conversion (`09` §3 decisions 16, 35, 37, 38; §6.4 part 10; v0.273.0) ─────
+//
+// WHAT IT IS FOR. The postgres image converts nothing: it REFUSES to start on an older major's datadir
+// (R-463, measured 2026-09-21: "database files are incompatible with server"), and 18 refuses even an
+// EMPTY volume at the old mount point (measured 2026-09-25, audits/night-2026-09-26/A/). So a step that
+// moves a PostgreSQL image across a major must rebuild the datadir: save everything from the OLD engine,
+// start the NEW one empty, load it back, check.
+//
+// THE BOX NEVER GUESSES. It converts only when the ladder entry of the step carries the harness's mark
+// (`engine_conversion: {service, engine, from, to}`), written by `upgrade-test.py --write-ladder` after
+// the product's own conversion was proven on a scratch box. A step that moves a PostgreSQL image across a
+// major WITHOUT the mark is refused before anything moves (defence in depth — the catalog gate should
+// never let one through). Pinned by TestConvert_MajorWithoutMarkIsRefused.
+//
+// IT RIDES THE GUARDED UPDATE. Precondition backup → safety dump → pin → pull → undo copy of every named
+// volume (the old datadir, with its finished-marker) → CONVERTING → start → verify with the new probe.
+// Any failure after the copy — the old engine will not start, the dump is cut off, the load errors, the
+// counts differ, the new datadir is the wrong major, the app is unhealthy — goes to the EXISTING undo:
+// every volume back from its copy, old pin, old definition, old probe. `undone`, else HOLD.
+//
+// THE ORDER INSIDE `converting`, and the order is the point:
+//
+//  1. the OLD database container — stopped by the copy, never recreated — is started ALONE;
+//  2. the check's "before": per database owner/encoding/collation, every role, every extension, every
+//     table's exact row count (decision 38 / A3);
+//  3. `pg_dumpall` from the old engine into the stack dir, validated by its completion line;
+//  4. the old engine stops;
+//  5. THE DB VOLUME IS EMPTIED ONLY AFTER THE COPY'S FINISHED-MARKER IS VALIDATED AGAIN — here, and again
+//     inside the helper that empties it (the Restore shape). Pinned by TestConvert_NeverEmptiedWithoutMarker;
+//  6. the NEW engine starts alone on the empty volume (`up -d --no-deps <svc>` with the new definition),
+//     its entrypoint initialising the bootstrap role and database from the app's own env;
+//  7. the entrypoint's databases are dropped when they hold no table, the dump's `CREATE ROLE <x>;` line is
+//     skipped for a role that already exists (its `ALTER ROLE … PASSWORD` still runs), and the load runs
+//     with ON_ERROR_STOP — so ANY other error stops it (measured: raw, exactly two "already exists");
+//  8. the check's "after" must EQUAL the before, and `$PGDATA/PG_VERSION` must be the mark's `to`.
+//
+// AFTER SUCCESS the old datadir's copy is KEPT (decision 35, B5) until the first backup of the converted
+// app is proven on the new engine; ReleaseConversionCopies then removes it, logged by name both times.
+
+// UpdatePhaseConverting is the conversion's phase, between copying and starting.
+const UpdatePhaseConverting = "converting"
+
+// preUpdateConvertDir holds the conversion's dump inside the stack dir. Kept on a HOLD (evidence),
+// removed on done and undone.
+const preUpdateConvertDir = "pre-update-convert"
+
+// EngineConversion is the ladder entry's mark: this step converts `service` from major `from` to `to`.
+type EngineConversion struct {
+	Service string `yaml:"service" json:"service"`
+	Engine  string `yaml:"engine" json:"engine"`
+	From    int    `yaml:"from" json:"from"`
+	To      int    `yaml:"to" json:"to"`
+}
+
+// ConversionCopy is app.yaml's record of the old datadir's copy kept after a successful conversion.
+type ConversionCopy struct {
+	Volume string `yaml:"volume" json:"volume"`
+	Copy   string `yaml:"copy" json:"copy"`
+	At     string `yaml:"at" json:"at"` // RFC3339 — a backup proven after this releases the copy
+	From   int    `yaml:"from" json:"from"`
+	To     int    `yaml:"to" json:"to"`
+}
+
+// conversionDumpMargin is A5's margin on the dump's bound (the DB volume's own size).
+const conversionDumpMargin = 1.25
+
+// ── recognising a PostgreSQL major move ──────────────────────────────────────────────────────────
+
+// isPostgresImage is the backup side's predicate (appbackup.dbTypeForImage, R-484) for the Postgres
+// family — kept equal by TestConvert_PostgresFamilyMatchesTheBackupSide.
+func isPostgresImage(ref string) bool {
+	img := strings.ToLower(ref)
+	return strings.Contains(img, "postgres") || strings.Contains(img, "postgis") ||
+		strings.Contains(img, "pgvector") || strings.Contains(img, "timescaledb")
+}
+
+var leadingDigits = regexp.MustCompile(`^(\d+)`)
+
+// postgresMajor reads the major from a Postgres-family tag: `16-alpine` → 16, `17.2` → 17,
+// `16-3.5-alpine` (postgis) → 16, `pg16` (pgvector) → 16, `16-vectorchord0.4.3-…` → 16.
+func postgresMajor(ref string) (int, bool) {
+	ref = stripDigest(ref)
+	_, tag := splitImageRef(ref)
+	tag = strings.TrimPrefix(strings.ToLower(tag), "pg")
+	m := leadingDigits.FindStringSubmatch(tag)
+	if m == nil {
+		return 0, false
+	}
+	n, err := strconv.Atoi(m[1])
+	return n, err == nil
+}
+
+func stripDigest(ref string) string {
+	if i := strings.Index(ref, "@"); i >= 0 {
+		return ref[:i]
+	}
+	return ref
+}
+
+// errNoConversionMark is the refusal of a PostgreSQL major move the ladder does not mark (B1).
+type errNoConversionMark struct{ service, from, to, why string }
+
+func (e *errNoConversionMark) Error() string {
+	return fmt.Sprintf("service %s moves PostgreSQL %s → %s across a major and %s — refusing before anything moves", e.service, e.from, e.to, e.why)
+}
+
+// planEngineConversion decides whether the step about to be pinned is a PostgreSQL major move, and if
+// so whether its ladder entry marks it. Returns nil, nil when no Postgres major moves.
+func planEngineConversion(pinned, stepImgs map[string]string, entry *LadderEntry) (*EngineConversion, error) {
+	var moved []string
+	for svc, newRef := range stepImgs {
+		oldRef, ok := pinned[svc]
+		if !ok || !isPostgresImage(newRef) {
+			continue
+		}
+		if stripDigest(oldRef) == stripDigest(newRef) {
+			continue
+		}
+		om, ok1 := postgresMajor(oldRef)
+		nm, ok2 := postgresMajor(newRef)
+		if ok1 && ok2 && om == nm {
+			continue // within a major — the engine reads its own datadir
+		}
+		moved = append(moved, svc)
+	}
+	if len(moved) == 0 {
+		return nil, nil
+	}
+	sort.Strings(moved)
+	svc := moved[0]
+	fail := func(why string) (*EngineConversion, error) {
+		return nil, &errNoConversionMark{service: svc, from: pinned[svc], to: stepImgs[svc], why: why}
+	}
+	if len(moved) > 1 {
+		return fail(fmt.Sprintf("so do %v — one conversion per step", moved[1:]))
+	}
+	om, ok1 := postgresMajor(pinned[svc])
+	nm, ok2 := postgresMajor(stepImgs[svc])
+	if !ok1 || !ok2 {
+		return fail("the major cannot be read from the tag")
+	}
+	if entry == nil || entry.EngineConversion == nil {
+		return fail("the step's ladder entry carries no engine_conversion mark (no test proved the conversion)")
+	}
+	c := entry.EngineConversion
+	if c.Service != svc || !strings.EqualFold(c.Engine, "postgres") || c.From != om || c.To != nm {
+		return fail(fmt.Sprintf("the mark says %s %s %d → %d, the step moves %s %d → %d", c.Service, c.Engine, c.From, c.To, svc, om, nm))
+	}
+	cp := *c
+	return &cp, nil
+}
+
+// ── the compose facts the conversion needs ──────────────────────────────────────────────────────
+
+type composeSvcVolumesDoc struct {
+	Name     string `yaml:"name"`
+	Services map[string]struct {
+		Volumes []interface{} `yaml:"volumes"`
+	} `yaml:"services"`
+	Volumes map[string]*composeVolDef `yaml:"volumes"`
+}
+
+// composeProject is the compose project name the manager runs a stack under: the file's top-level
+// `name:`, else the stack directory's name (DeclaredVolumeNames' rule).
+func composeProject(composePath string) string {
+	data, err := os.ReadFile(composePath)
+	if err == nil {
+		var doc composeSvcVolumesDoc
+		if yaml.Unmarshal(data, &doc) == nil && strings.TrimSpace(doc.Name) != "" {
+			return strings.TrimSpace(doc.Name)
+		}
+	}
+	return filepath.Base(filepath.Dir(composePath))
+}
+
+// serviceDataVolume returns the declared named volume a service mounts at /var/lib/postgresql or below:
+// its compose KEY and its Docker name.
+func serviceDataVolume(composePath, service string) (key, dockerName string, err error) {
+	data, err := os.ReadFile(composePath)
+	if err != nil {
+		return "", "", err
+	}
+	var doc composeSvcVolumesDoc
+	if err := yaml.Unmarshal(data, &doc); err != nil {
+		return "", "", fmt.Errorf("parsing %s: %w", composePath, err)
+	}
+	svc, ok := doc.Services[service]
+	if !ok {
+		return "", "", fmt.Errorf("%s declares no service %q", composePath, service)
+	}
+	project := strings.TrimSpace(doc.Name)
+	if project == "" {
+		project = filepath.Base(filepath.Dir(composePath))
+	}
+	for _, v := range svc.Volumes {
+		src, dst := "", ""
+		switch x := v.(type) {
+		case string:
+			parts := strings.Split(x, ":")
+			if len(parts) >= 2 {
+				src, dst = parts[0], parts[1]
+			}
+		case map[string]interface{}:
+			src, _ = x["source"].(string)
+			dst, _ = x["target"].(string)
+		}
+		if !strings.HasPrefix(dst, "/var/lib/postgresql") {
+			continue
+		}
+		def, declared := doc.Volumes[src]
+		if !declared {
+			return "", "", fmt.Errorf("service %s mounts %s at %s, which is not a named volume the file declares (a bind mount cannot be converted)", service, src, dst)
+		}
+		name := project + "_" + src
+		if def != nil && strings.TrimSpace(def.Name) != "" {
+			name = strings.TrimSpace(def.Name)
+		}
+		return src, name, nil
+	}
+	return "", "", fmt.Errorf("service %s mounts no named volume under /var/lib/postgresql", service)
+}
+
+// freeBytesAt is the free space of the filesystem holding path (the dump's home). Seam for tests.
+func (m *Manager) freeBytesAt(path string) (int64, bool) {
+	if m.convertFreeFn != nil {
+		return m.convertFreeFn(path)
+	}
+	var s syscall.Statfs_t
+	if err := syscall.Statfs(path, &s); err != nil {
+		return 0, false
+	}
+	return int64(s.Bavail) * int64(s.Bsize), true
+}
+
+// convertSpaceError is the refusal of B6's „no-space" sentence.
+type convertSpaceError struct{ need, free int64 }
+
+func (e *convertSpaceError) Error() string {
+	return fmt.Sprintf("the conversion needs %s free for its dump (the database volume's size × %.2f + the %.0f GB floor) and %s is free", humanBytes(e.need), conversionDumpMargin, updateDiskFloorGiB, humanBytes(e.free))
+}
+
+func humanBytes(b int64) string {
+	switch {
+	case b >= 1<<30:
+		return fmt.Sprintf("%.1f GB", float64(b)/(1<<30))
+	case b >= 1<<20:
+		return fmt.Sprintf("%.0f MB", float64(b)/(1<<20))
+	}
+	return fmt.Sprintf("%d kB", b/1024)
+}
+
+// planConversionSpace checks, before anything moves, that the conversion's dump fits beside the stack
+// dir (A5): the dump's bound is the DB volume's own size, with a margin, plus the update's 2 GB floor.
+// The undo copy's own space is planUndoCopies' check, unchanged.
+func (m *Manager) planConversionSpace(name, dir, dbVol string) error {
+	b, err := m.copier().VolumeBytes(dbVol)
+	if err != nil {
+		return fmt.Errorf("sizing the database volume %s: %w", dbVol, err)
+	}
+	need := int64(float64(b)*conversionDumpMargin) + int64(updateDiskFloorGiB*(1<<30))
+	free, known := m.freeBytesAt(dir)
+	if !known {
+		m.logger.Printf("[WARN] [stacks] update %s: free space beside the stack dir is unreadable — the conversion proceeds without its space check", name)
+		return nil
+	}
+	m.logger.Printf("[INFO] [stacks] update %s: conversion space — database volume %s is %s; the dump needs up to %s + the %.0f GB floor; %s free beside the stack dir", name, dbVol, humanBytes(b), humanBytes(int64(float64(b)*conversionDumpMargin)), updateDiskFloorGiB, humanBytes(free))
+	if free < need {
+		return &convertSpaceError{need: need, free: free}
+	}
+	return nil
+}
+
+// ── the process boundary ────────────────────────────────────────────────────────────────────────
+
+// pgConverter is the conversion's process boundary. Production is dockerPGConverter; tests inject a
+// fake and never touch docker (R-650).
+type pgConverter interface {
+	ServiceContainer(project, service string) (string, error)
+	Start(container string) error
+	Stop(container string) error
+	Env(container, key string) string
+	WaitReady(ctx context.Context, container, user string, timeout time.Duration) error
+	Snapshot(container, user string) ([]string, error)
+	DumpAll(container, user, path string) error
+	// Empty empties vol ONLY when copyVol carries its finished-marker (checked inside the same helper).
+	Empty(copyVol, vol string) error
+	// DropEmptyDatabases drops every non-template database except `postgres` that holds no table, and
+	// refuses (error) if any holds one — only an entrypoint-made database may be dropped.
+	DropEmptyDatabases(container, user string) ([]string, error)
+	Roles(container, user string) ([]string, error)
+	Load(container, user, path string, skipLines map[string]bool) error
+	DataVersion(container string) (string, error)
+}
+
+func (m *Manager) converter() pgConverter {
+	if m.pgConv != nil {
+		return m.pgConv
+	}
+	return dockerPGConverter{m: m}
+}
+
+// ── the conversion ──────────────────────────────────────────────────────────────────────────────
+
+// dumpAllCompleteLine is the last comment pg_dumpall writes; a dump without it is cut off.
+const dumpAllCompleteLine = "PostgreSQL database cluster dump complete"
+
+// validateDumpAll refuses a dump without its completion line (the lesson of 2026-09-23: a truncated
+// PostgreSQL dump loads with exit 0).
+func validateDumpAll(path string) error {
+	f, err := os.Open(path)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	st, err := f.Stat()
+	if err != nil {
+		return err
+	}
+	off := st.Size() - 4096
+	if off < 0 {
+		off = 0
+	}
+	buf := make([]byte, st.Size()-off)
+	if _, err := f.ReadAt(buf, off); err != nil && err != io.EOF {
+		return err
+	}
+	if !strings.Contains(string(buf), dumpAllCompleteLine) {
+		return fmt.Errorf("the dump %s (%d bytes) has no completion line — it is cut off", path, st.Size())
+	}
+	return nil
+}
+
+// convertEngine is the `converting` phase. On error the caller runs the undo (failAndHold → tryUndo).
+func (m *Manager) convertEngine(ctx context.Context, name, dir string, env []string, entry *updateJournalEntry) error {
+	c := entry.Convert
+	pc := m.converter()
+	t0 := m.now()
+	project := composeProject(ComposePathIn(dir))
+	dbVol := entry.ConvertVolume
+	var copyVol string
+	for _, u := range entry.UndoCopies {
+		if u.Volume == dbVol {
+			copyVol = u.Copy
+		}
+	}
+	if copyVol == "" {
+		return fmt.Errorf("the database volume %s has no undo copy — refusing to touch it", dbVol)
+	}
+	m.logger.Printf("[INFO] [stacks] update %s: CONVERTING %s PostgreSQL %d → %d (volume %s, kept copy %s)", name, c.Service, c.From, c.To, dbVol, copyVol)
+
+	// 1. the OLD engine, alone. The copy stopped the app's containers without recreating them, so the
+	//    service's container is still the old image.
+	oldC, err := pc.ServiceContainer(project, c.Service)
+	if err != nil {
+		return fmt.Errorf("finding the old database container: %w", err)
+	}
+	if err := pc.Start(oldC); err != nil {
+		return fmt.Errorf("starting the old engine %s: %w", oldC, err)
+	}
+	user := pc.Env(oldC, "POSTGRES_USER")
+	if user == "" {
+		user = "postgres"
+	}
+	if err := pc.WaitReady(ctx, oldC, user, 3*time.Minute); err != nil {
+		_ = pc.Stop(oldC)
+		return fmt.Errorf("the old engine did not become ready: %w", err)
+	}
+	// 2. the check's "before"
+	before, err := pc.Snapshot(oldC, user)
+	if err != nil {
+		_ = pc.Stop(oldC)
+		return fmt.Errorf("reading the old engine's counts: %w", err)
+	}
+	// 3. the dump, validated
+	cdir := filepath.Join(dir, preUpdateConvertDir)
+	if err := os.MkdirAll(cdir, 0o700); err != nil {
+		_ = pc.Stop(oldC)
+		return fmt.Errorf("creating %s: %w", cdir, err)
+	}
+	dump := filepath.Join(cdir, c.Service+"-dumpall.sql")
+	td := time.Now()
+	if err := pc.DumpAll(oldC, user, dump); err != nil {
+		_ = pc.Stop(oldC)
+		return fmt.Errorf("pg_dumpall from the old engine: %w", err)
+	}
+	if err := validateDumpAll(dump); err != nil {
+		_ = pc.Stop(oldC)
+		return err
+	}
+	var dumpSize int64
+	if fi, err := os.Stat(dump); err == nil {
+		dumpSize = fi.Size()
+	}
+	m.logger.Printf("[INFO] [stacks] update %s: pg_dumpall from %d done in %s — %s, completion line present; before: %s", name, c.From, time.Since(td).Round(time.Millisecond), humanBytes(dumpSize), summariseSnapshot(before))
+	// 4. the old engine stops
+	if err := pc.Stop(oldC); err != nil {
+		return fmt.Errorf("stopping the old engine: %w", err)
+	}
+	// 5. the volume is emptied ONLY after the copy's marker is validated again
+	if !m.copier().Complete(copyVol) {
+		return fmt.Errorf("the kept copy %s has no finished-marker — the database volume is NOT emptied", copyVol)
+	}
+	entry.ConvertTouched = true
+	if !m.enterUpdatePhase(name, entry, UpdatePhaseConverting) {
+		return fmt.Errorf("could not journal that the database volume is about to be emptied")
+	}
+	if err := pc.Empty(copyVol, dbVol); err != nil {
+		return fmt.Errorf("emptying %s: %w", dbVol, err)
+	}
+	m.logger.Printf("[INFO] [stacks] update %s: database volume %s emptied (its copy %s holds the old datadir, marker checked twice)", name, dbVol, copyVol)
+	// 6. the NEW engine, alone, on the empty volume
+	if _, err := m.updateCompose(dir, env, "up", "-d", "--no-deps", c.Service); err != nil {
+		return fmt.Errorf("starting the new engine: %w", err)
+	}
+	newC, err := pc.ServiceContainer(project, c.Service)
+	if err != nil {
+		return fmt.Errorf("finding the new database container: %w", err)
+	}
+	if err := pc.WaitReady(ctx, newC, user, 5*time.Minute); err != nil {
+		return fmt.Errorf("the new engine did not become ready: %w", err)
+	}
+	// 7. make room for exactly the two objects the entrypoint made, then load with ON_ERROR_STOP
+	dropped, err := pc.DropEmptyDatabases(newC, user)
+	if err != nil {
+		return fmt.Errorf("preparing the new engine: %w", err)
+	}
+	roles, err := pc.Roles(newC, user)
+	if err != nil {
+		return fmt.Errorf("reading the new engine's roles: %w", err)
+	}
+	skip := map[string]bool{}
+	for _, r := range roles {
+		skip["CREATE ROLE "+quoteIdent(r)+";"] = true
+	}
+	tl := time.Now()
+	if err := pc.Load(newC, user, dump, skip); err != nil {
+		return fmt.Errorf("loading the dump into %d: %w", c.To, err)
+	}
+	m.logger.Printf("[INFO] [stacks] update %s: loaded into %d in %s (dropped the entrypoint's empty database(s) %v; skipped CREATE ROLE for %v)", name, c.To, time.Since(tl).Round(time.Millisecond), dropped, roles)
+	// 8. the check's "after" must equal the before; PG_VERSION must be the mark's `to`
+	after, err := pc.Snapshot(newC, user)
+	if err != nil {
+		return fmt.Errorf("reading the new engine's counts: %w", err)
+	}
+	if diff := diffSnapshots(before, after); diff != "" {
+		return fmt.Errorf("the check differs after the load: %s", diff)
+	}
+	ver, err := pc.DataVersion(newC)
+	if err != nil {
+		return fmt.Errorf("reading PG_VERSION of the new datadir: %w", err)
+	}
+	if strings.TrimSpace(ver) != strconv.Itoa(c.To) {
+		return fmt.Errorf("the new datadir's PG_VERSION is %q, the mark says %d", strings.TrimSpace(ver), c.To)
+	}
+	m.logger.Printf("[INFO] [stacks] update %s: CONVERTED %d → %d in %s — the check is equal (%s), PG_VERSION %s", name, c.From, c.To, m.now().Sub(t0).Round(time.Millisecond), summariseSnapshot(after), strings.TrimSpace(ver))
+	return nil
+}
+
+func quoteIdent(s string) string {
+	if regexp.MustCompile(`^[a-z_][a-z0-9_$]*$`).MatchString(s) && !pgReserved[s] {
+		return s
+	}
+	return `"` + strings.ReplaceAll(s, `"`, `""`) + `"`
+}
+
+// pgReserved is the handful of reserved words a role could plausibly be named (quote_ident quotes them).
+var pgReserved = map[string]bool{"user": true, "all": true, "default": true, "group": true, "order": true, "table": true}
+
+// summariseSnapshot is the log's one-line form: databases, tables, rows.
+func summariseSnapshot(lines []string) string {
+	dbs, tables, rows := 0, 0, int64(0)
+	for _, l := range lines {
+		switch {
+		case strings.HasPrefix(l, "db:"):
+			dbs++
+		case strings.HasPrefix(l, "rows:"):
+			tables++
+			if i := strings.LastIndex(l, "="); i >= 0 {
+				n, _ := strconv.ParseInt(l[i+1:], 10, 64)
+				rows += n
+			}
+		}
+	}
+	return fmt.Sprintf("%d database(s), %d table(s), %d row(s)", dbs, tables, rows)
+}
+
+// diffSnapshots names the first differences, "" when equal.
+func diffSnapshots(before, after []string) string {
+	a, b := map[string]bool{}, map[string]bool{}
+	for _, l := range before {
+		a[l] = true
+	}
+	for _, l := range after {
+		b[l] = true
+	}
+	var d []string
+	for _, l := range before {
+		if !b[l] {
+			d = append(d, "missing after: "+l)
+		}
+	}
+	for _, l := range after {
+		if !a[l] {
+			d = append(d, "new after: "+l)
+		}
+	}
+	if len(d) == 0 {
+		return ""
+	}
+	if len(d) > 6 {
+		d = append(d[:6], fmt.Sprintf("… %d more", len(d)-6))
+	}
+	return strings.Join(d, "; ")
+}
+
+// ── keeping, then releasing, the old datadir's copy (B5) ────────────────────────────────────────
+
+func (m *Manager) recordConversionCopy(name, dir string, cc *ConversionCopy) {
+	cfg := LoadAppConfig(dir)
+	if cfg == nil || (cc == nil && cfg.ConversionCopy == nil) {
+		return
+	}
+	cfg.ConversionCopy = cc
+	meta := LoadMetadata(dir)
+	if err := SaveAppConfig(dir, cfg, m.encKey, SensitiveEnvVars(&meta)); err != nil {
+		m.logger.Printf("[ERROR] [stacks] update %s: recording conversion_copy failed: %v", name, err)
+		return
+	}
+	m.mu.Lock()
+	if st, ok := m.stacks[name]; ok && st.AppConfig != nil {
+		st.AppConfig.ConversionCopy = cc
+	}
+	m.mu.Unlock()
+}
+
+// ReleaseConversionCopies removes each kept pre-conversion datadir copy whose app now has a backup
+// PROVEN after the conversion (any tier — the backup side returns only proven copies). Returns the
+// names released. Run periodically from main.go (TestConvert_ReleaseIsWiredAtStartup).
+func (m *Manager) ReleaseConversionCopies(ctx context.Context) []string {
+	g := m.guards()
+	if g == nil {
+		return nil
+	}
+	var released []string
+	for _, st := range m.GetStacks() {
+		if st.AppConfig == nil || st.AppConfig.ConversionCopy == nil {
+			continue
+		}
+		cc := st.AppConfig.ConversionCopy
+		at, err := time.Parse(time.RFC3339, cc.At)
+		if err != nil {
+			m.logger.Printf("[WARN] [stacks] %s: conversion_copy has an unreadable time %q — the copy %s is kept", st.Name, cc.At, cc.Copy)
+			continue
+		}
+		rp, ok, _ := g.RestorePoints(ctx, st.Name, func(p UpdateRestorePoint) bool { return p.ProvenAt.After(at) })
+		if !ok {
+			continue
+		}
+		if err := m.copier().Remove(cc.Copy); err != nil {
+			m.logger.Printf("[WARN] [stacks] %s: could not remove the pre-conversion copy %s: %v — kept, tried again later", st.Name, cc.Copy, err)
+			continue
+		}
+		m.recordConversionCopy(st.Name, filepath.Dir(st.ComposePath), nil)
+		m.logger.Printf("[INFO] [stacks] %s: REMOVED the pre-conversion datadir copy %s (PostgreSQL %d) — the converted app has a backup proven on %d: %s at %s", st.Name, cc.Copy, cc.From, cc.To, updateTierName(rp.Tier), rp.ProvenAt.UTC().Format(time.RFC3339))
+		released = append(released, st.Name)
+	}
+	return released
+}
+
+// ── production boundary ─────────────────────────────────────────────────────────────────────────
+
+type dockerPGConverter struct{ m *Manager }
+
+func (d dockerPGConverter) ServiceContainer(project, service string) (string, error) {
+	out, err := d.m.execCommand("docker", "ps", "-a", "--filter", "label=com.docker.compose.project="+project,
+		"--filter", "label=com.docker.compose.service="+service, "--format", "{{.Names}}")
+	if err != nil {
+		return "", err
+	}
+	var names []string
+	for _, l := range strings.Split(out, "\n") {
+		if l = strings.TrimSpace(l); l != "" {
+			names = append(names, l)
+		}
+	}
+	if len(names) != 1 {
+		return "", fmt.Errorf("project %s service %s has %d container(s) %v, want exactly one", project, service, len(names), names)
+	}
+	return names[0], nil
+}
+
+func (d dockerPGConverter) Start(c string) error {
+	_, err := d.m.execCommand("docker", "start", c)
+	return err
+}
+func (d dockerPGConverter) Stop(c string) error {
+	_, err := d.m.execCommand("docker", "stop", "-t", "60", c)
+	return err
+}
+
+func (d dockerPGConverter) Env(c, key string) string {
+	out, err := d.m.execCommand("docker", "inspect", c, "--format", "{{range .Config.Env}}{{println .}}{{end}}")
+	if err != nil {
+		return ""
+	}
+	for _, l := range strings.Split(out, "\n") {
+		if strings.HasPrefix(l, key+"=") {
+			return strings.TrimSpace(strings.TrimPrefix(l, key+"="))
+		}
+	}
+	return ""
+}
+
+// WaitReady asks over TCP on 127.0.0.1: the entrypoint's temporary init server listens on the socket
+// only, so a TCP answer is the FINAL server (loading into the temporary one would be lost on its stop).
+func (d dockerPGConverter) WaitReady(ctx context.Context, c, user string, timeout time.Duration) error {
+	deadline := time.Now().Add(timeout)
+	last := ""
+	for time.Now().Before(deadline) {
+		out, err := d.m.execCommand("docker", "exec", c, "psql", "-h", "127.0.0.1", "-U", user, "-d", "postgres", "-Atc", "select 1")
+		if err == nil && strings.TrimSpace(out) == "1" {
+			return nil
+		}
+		if err != nil {
+			last = err.Error()
+		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(2 * time.Second):
+		}
+	}
+	return fmt.Errorf("no answer within %s (last: %s)", timeout, firstLine(last))
+}
+
+func firstLine(s string) string {
+	if i := strings.IndexByte(s, '\n'); i >= 0 {
+		return s[:i]
+	}
+	return s
+}
+
+func (d dockerPGConverter) psql(c, user, db, sql string) (string, error) {
+	return d.m.execCommand("docker", "exec", c, "psql", "-h", "127.0.0.1", "-U", user, "-d", db, "-v", "ON_ERROR_STOP=1", "-Atc", sql)
+}
+
+const snapshotCountsSQL = `SELECT n.nspname||'.'||c.relname||'='||(xpath('/row/c/text()', query_to_xml(format('select count(*) as c from %I.%I', n.nspname, c.relname), false, true, '')))[1]::text FROM pg_class c JOIN pg_namespace n ON n.oid=c.relnamespace WHERE c.relkind IN ('r','p') AND n.nspname NOT IN ('pg_catalog','information_schema') AND n.nspname NOT LIKE 'pg_toast%' ORDER BY 1`
+
+// Snapshot is A3's check, as measured on 9202.
+func (d dockerPGConverter) Snapshot(c, user string) ([]string, error) {
+	var out []string
+	add := func(prefix, s string) {
+		for _, l := range strings.Split(s, "\n") {
+			if l = strings.TrimSpace(l); l != "" {
+				out = append(out, prefix+l)
+			}
+		}
+	}
+	s, err := d.psql(c, user, "postgres", "select datname||' owner='||pg_get_userbyid(datdba)||' enc='||pg_encoding_to_char(encoding)||' coll='||datcollate from pg_database where not datistemplate order by 1")
+	if err != nil {
+		return nil, err
+	}
+	add("db:", s)
+	s, err = d.psql(c, user, "postgres", "select rolname||' super='||rolsuper||' login='||rolcanlogin||' pw='||(rolpassword is not null) from pg_roles where rolname !~ '^pg_' order by 1")
+	if err != nil {
+		return nil, err
+	}
+	add("role:", s)
+	dbs, err := d.psql(c, user, "postgres", "select datname from pg_database where not datistemplate and datname<>'postgres' order by 1")
+	if err != nil {
+		return nil, err
+	}
+	for _, db := range strings.Split(dbs, "\n") {
+		if db = strings.TrimSpace(db); db == "" {
+			continue
+		}
+		s, err := d.psql(c, user, db, "select extname from pg_extension order by 1")
+		if err != nil {
+			return nil, err
+		}
+		add("ext:"+db+":", s)
+		s, err = d.psql(c, user, db, snapshotCountsSQL)
+		if err != nil {
+			return nil, err
+		}
+		add("rows:"+db+":", s)
+	}
+	sort.Strings(out)
+	return out, nil
+}
+
+// DumpAll streams pg_dumpall's stdout straight into the file — never through the controller's memory.
+func (d dockerPGConverter) DumpAll(c, user, path string) error {
+	f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600)
+	if err != nil {
+		return err
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Hour)
+	defer cancel()
+	cmd := dockerexec.CommandContext(ctx, "docker", "exec", c, "pg_dumpall", "-h", "127.0.0.1", "-U", user)
+	var stderr strings.Builder
+	cmd.Stdout, cmd.Stderr = f, &stderr
+	runErr := cmd.Run()
+	if err := f.Sync(); err != nil && runErr == nil {
+		runErr = err
+	}
+	if err := f.Close(); err != nil && runErr == nil {
+		runErr = err
+	}
+	if runErr != nil {
+		return fmt.Errorf("%w (stderr: %s)", runErr, firstLine(stderr.String()))
+	}
+	return nil
+}
+
+func (d dockerPGConverter) Empty(copyVol, vol string) error {
+	_, err := d.m.execCommand("docker", "run", "--rm", "-v", copyVol+":/from:ro", "-v", vol+":/to", undoHelperImage,
+		"sh", "-c", "test -f /from/"+undoCopyMarker+" && find /to -mindepth 1 -delete && sync")
+	return err
+}
+
+func (d dockerPGConverter) DropEmptyDatabases(c, user string) ([]string, error) {
+	dbs, err := d.psql(c, user, "postgres", "select datname from pg_database where not datistemplate and datname<>'postgres' order by 1")
+	if err != nil {
+		return nil, err
+	}
+	var dropped []string
+	for _, db := range strings.Split(dbs, "\n") {
+		if db = strings.TrimSpace(db); db == "" {
+			continue
+		}
+		n, err := d.psql(c, user, db, "select count(*) from pg_class c join pg_namespace n on n.oid=c.relnamespace where c.relkind in ('r','p') and n.nspname not in ('pg_catalog','information_schema')")
+		if err != nil {
+			return nil, err
+		}
+		if strings.TrimSpace(n) != "0" {
+			return nil, fmt.Errorf("database %s on the NEW engine already holds %s table(s) — it was not made empty by the entrypoint; refusing to drop it", db, strings.TrimSpace(n))
+		}
+		if _, err := d.psql(c, user, "postgres", "DROP DATABASE "+quoteIdent(db)); err != nil {
+			return nil, err
+		}
+		dropped = append(dropped, db)
+	}
+	return dropped, nil
+}
+
+func (d dockerPGConverter) Roles(c, user string) ([]string, error) {
+	s, err := d.psql(c, user, "postgres", "select rolname from pg_roles where rolname !~ '^pg_' order by 1")
+	if err != nil {
+		return nil, err
+	}
+	var out []string
+	for _, l := range strings.Split(s, "\n") {
+		if l = strings.TrimSpace(l); l != "" {
+			out = append(out, l)
+		}
+	}
+	return out, nil
+}
+
+// Load streams the dump into psql with ON_ERROR_STOP, skipping exactly the lines named.
+func (d dockerPGConverter) Load(c, user, path string, skip map[string]bool) error {
+	f, err := os.Open(path)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	pr, pw := io.Pipe()
+	go func() {
+		pw.CloseWithError(filterLines(f, pw, skip))
+	}()
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Hour)
+	defer cancel()
+	cmd := dockerexec.CommandContext(ctx, "docker", "exec", "-i", c, "psql", "-h", "127.0.0.1", "-U", user, "-d", "postgres", "-v", "ON_ERROR_STOP=1", "-q")
+	var stderr strings.Builder
+	cmd.Stdin, cmd.Stderr, cmd.Stdout = pr, &stderr, io.Discard
+	if err := cmd.Run(); err != nil {
+		return fmt.Errorf("%w (psql: %s)", err, firstLine(strings.TrimSpace(stderr.String())))
+	}
+	return nil
+}
+
+// filterLines copies r to w, dropping each line that equals a key of skip exactly.
+func filterLines(r io.Reader, w io.Writer, skip map[string]bool) error {
+	br := bufio.NewReaderSize(r, 1<<20)
+	for {
+		line, err := br.ReadString('\n')
+		if len(line) > 0 && !skip[strings.TrimRight(line, "\r\n")] {
+			if _, werr := io.WriteString(w, line); werr != nil {
+				return werr
+			}
+		}
+		if err == io.EOF {
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+	}
+}
+
+func (d dockerPGConverter) DataVersion(c string) (string, error) {
+	return d.m.execCommand("docker", "exec", c, "sh", "-c", `cat "$PGDATA/PG_VERSION"`)
+}
+
+// appLabel is the app's display name for a household sentence, else its stack name.
+func appLabel(st *Stack) string {
+	if st != nil && strings.TrimSpace(st.Meta.DisplayName) != "" {
+		return st.Meta.DisplayName
+	}
+	if st == nil {
+		return ""
+	}
+	return st.Name
+}
+
+// conversionFor is B1's decision for one step, shared by the preflight and the job: the conversion to
+// run (nil = none), the DB volume's Docker name, and on refusal the bundle key that names it.
+func (m *Manager) conversionFor(st *Stack, pinned map[string]string, step LadderStep) (*EngineConversion, string, string, error) {
+	imgs, err := ParseComposeImages(step.Source)
+	if err != nil {
+		// Not this check's refusal: advancePinTo reads the same file and refuses with its own sentence.
+		return nil, "", "", nil
+	}
+	conv, err := planEngineConversion(pinned, imgs, step.Entry)
+	if err != nil {
+		return nil, "", "err.stacks.update_engine_no_test", err
+	}
+	if conv == nil {
+		return nil, "", "", nil
+	}
+	oldKey, vol, err := serviceDataVolume(st.ComposePath, conv.Service)
+	if err != nil {
+		return nil, "", "", fmt.Errorf("the running definition: %w", err)
+	}
+	newKey, _, err := serviceDataVolume(step.Source, conv.Service)
+	if err != nil {
+		return nil, "", "", fmt.Errorf("the step's definition: %w", err)
+	}
+	if oldKey != newKey {
+		return nil, "", "", fmt.Errorf("the step mounts volume %q for %s, the running definition %q — the conversion rebuilds the datadir IN its volume and the undo copies that one", newKey, conv.Service, oldKey)
+	}
+	return conv, vol, "", nil
+}

controller/internal/stacks/pgconvert_test.go

@@ -0,0 +1,507 @@
+package stacks
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+// v0.273.0 — the PostgreSQL major conversion (`09` §6.4 part 10, pgconvert.go). Every test runs the REAL
+// guarded update job with the process boundaries faked (compose, the volume copier, the PostgreSQL
+// converter, the health waits) and reads the EFFECT back: the bytes in the fake database volume, the pin,
+// the phase, the hold, the kept copy. Nothing here reaches Docker (R-650).
+
+const (
+	convOld     = "services:\n  web:\n    image: docmost/docmost:0.96.0\n  db:\n    image: postgres:16-alpine\n    volumes:\n      - pgdata:/var/lib/postgresql/data\nvolumes:\n  pgdata:\n"
+	convNew     = "services:\n  web:\n    image: docmost/docmost:0.96.0\n  db:\n    image: postgres:18-alpine\n    volumes:\n      - pgdata:/var/lib/postgresql\nvolumes:\n  pgdata:\n"
+	convVol     = "nextcloud_pgdata" // project = the stack dir's name (newPinManager names it nextcloud)
+	convPin     = "deployed: true\nenv: {}\npinned_images:\n    web: docmost/docmost:0.96.0\n    db: postgres:16-alpine\n"
+	convOldData = "PG16-DATADIR"
+	convNewData = "PG18-DATADIR"
+)
+
+func convLadder(mark string) string {
+	e := `  - {"from": {"web": "docmost/docmost:0.96.0", "db": "postgres:16-alpine"}, "to": {"web": "docmost/docmost:0.96.0", "db": "postgres:18-alpine"}, "digest": {}, "verdict": "proven"`
+	if mark != "" {
+		e += `, "engine_conversion": ` + mark
+	}
+	return "update_ladder:\n" + e + "}\n"
+}
+
+const goodMark = `{"service": "db", "engine": "postgres", "from": 16, "to": 18}`
+
+// fakePG is the PostgreSQL boundary. It works on the fake copier's volume content, so "the data came
+// back" is a string compare.
+type fakePG struct {
+	mu        sync.Mutex
+	fc        *fakeCopier
+	calls     []string
+	before    []string
+	after     []string // nil = equal to before
+	loadErr   error
+	dumpCut   bool
+	version   string
+	startErr  error
+	onConvert func(step string) // a hook to act between steps (the restart test)
+}
+
+func (p *fakePG) note(s string) {
+	p.mu.Lock()
+	p.calls = append(p.calls, s)
+	p.mu.Unlock()
+	if p.onConvert != nil {
+		p.onConvert(s)
+	}
+}
+func (p *fakePG) ServiceContainer(project, service string) (string, error) {
+	return project + "-" + service, nil
+}
+func (p *fakePG) Start(c string) error                                           { p.note("start " + c); return p.startErr }
+func (p *fakePG) Stop(c string) error                                            { p.note("stop " + c); return nil }
+func (p *fakePG) Env(string, string) string                                      { return "docmost" }
+func (p *fakePG) WaitReady(context.Context, string, string, time.Duration) error { return nil }
+func (p *fakePG) Snapshot(c, user string) ([]string, error) {
+	if p.fc.vol(convVol) == convOldData {
+		p.note("snapshot old")
+		return p.before, nil
+	}
+	p.note("snapshot new")
+	if p.after != nil {
+		return p.after, nil
+	}
+	return p.before, nil
+}
+func (p *fakePG) DumpAll(c, user, path string) error {
+	p.note("dumpall")
+	body := "CREATE ROLE docmost;\nALTER ROLE docmost WITH SUPERUSER;\n-- data\n"
+	if !p.dumpCut {
+		body += "--\n-- " + dumpAllCompleteLine + "\n--\n"
+	}
+	return os.WriteFile(path, []byte(body), 0o600)
+}
+func (p *fakePG) Empty(copyVol, vol string) error {
+	p.note("empty " + vol)
+	if !p.fc.Complete(copyVol) {
+		return fmt.Errorf("no marker in %s", copyVol)
+	}
+	p.fc.setVol(vol, "")
+	return nil
+}
+func (p *fakePG) DropEmptyDatabases(string, string) ([]string, error) {
+	p.note("drop-empty")
+	return []string{"docmost"}, nil
+}
+func (p *fakePG) Roles(string, string) ([]string, error) { return []string{"docmost"}, nil }
+func (p *fakePG) Load(c, user, path string, skip map[string]bool) error {
+	p.note("load")
+	if !skip["CREATE ROLE docmost;"] {
+		return fmt.Errorf("the existing role's CREATE line was not skipped")
+	}
+	if p.loadErr != nil {
+		return p.loadErr
+	}
+	p.fc.setVol(convVol, convNewData)
+	return nil
+}
+func (p *fakePG) DataVersion(string) (string, error) {
+	if p.version != "" {
+		return p.version, nil
+	}
+	return "18\n", nil
+}
+func (p *fakePG) called(prefix string) bool {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	for _, c := range p.calls {
+		if strings.HasPrefix(c, prefix) {
+			return true
+		}
+	}
+	return false
+}
+
+// convManager: slice 4's shape with docmost on postgres:16-alpine and the catalog offering 18 through a
+// one-entry ladder carrying `mark` ("" = no mark). The new version is healthy unless newHealthy is false;
+// the old one always answers its old probe.
+func convManager(t *testing.T, mark string) (*Manager, string, *fakeGuards, *composeRec, *fakeCopier, *fakePG) {
+	t.Helper()
+	m, dir := newPinManager(t, convOld, convNew, convPin)
+	mustWrite(t, AppliedComposePath(dir), convOld)
+	catDir := filepath.Dir(m.CatalogTemplatePath("nextcloud", "docker-compose.yml"))
+	mustWrite(t, filepath.Join(catDir, ".felhom.yml"), "display_name: Docmost\n"+convLadder(mark))
+	mustWrite(t, filepath.Join(dir, ".felhom.yml"), "display_name: Docmost\n")
+	m.stacks["nextcloud"].Meta = Metadata{DisplayName: "Docmost"}
+	g := &fakeGuards{points: []UpdateRestorePoint{{Tier: UpdateTierSecondDrive, ProvenAt: slice4T0.Add(-1 * time.Hour)}}, stackDir: dir}
+	c := &composeRec{fail: map[string]error{}}
+	m.updateGuards = g
+	m.updateComposeFn = c.fn
+	m.updateHealthFn = func(context.Context, string, time.Duration) (bool, string) { return true, "fake healthy" }
+	m.updateUndoHealthFn = func(context.Context, string, time.Duration, *Metadata) (bool, string) { return true, "old answered" }
+	m.updateMemoryFn = func(int, int, int, int) (error, string) { return nil, "" }
+	m.updateDiskFreeFn = func() (float64, bool) { return 50, true }
+	m.convertFreeFn = func(string) (int64, bool) { return 50 << 30, true }
+	m.updateNowFn = func() time.Time { return slice4T0 }
+	m.execFn = func(string, ...string) (string, error) { return "", nil }
+	fc := newFakeCopier(map[string]string{convVol: convOldData})
+	m.undoCopier = fc
+	pg := &fakePG{fc: fc, before: []string{"db:docmost owner=docmost", "role:docmost super=true", "ext:docmost:pg_trgm", "rows:docmost:public.users=1"}}
+	m.pgConv = pg
+	return m, dir, g, c, fc, pg
+}
+
+func dbPin(t *testing.T, dir string) string {
+	t.Helper()
+	return LoadAppConfig(dir).PinnedImages["db"]
+}
+
+// TestConvert_HappyPath — one press converts 16 → 18: the volume holds the new datadir, the pin names 18,
+// the phase passed through `converting`, the OLD datadir's copy is KEPT and recorded, the dump is gone.
+func TestConvert_HappyPath(t *testing.T) {
+	m, dir, _, c, fc, pg := convManager(t, goodMark)
+	if err := m.StartGuardedUpdate("nextcloud"); err != nil {
+		t.Fatal(err)
+	}
+	st := waitUpdateDone(t, m, "nextcloud")
+	if st.UpdatePhase != UpdatePhaseDone {
+		t.Fatalf("ended %q (%s)", st.UpdatePhase, st.UpdateError)
+	}
+	if got := fc.vol(convVol); got != convNewData {
+		t.Fatalf("the database volume holds %q, want the converted datadir", got)
+	}
+	if got := dbPin(t, dir); got != "postgres:18-alpine" {
+		t.Fatalf("pin %q, want postgres:18-alpine", got)
+	}
+	order := strings.Join(pg.calls, ",")
+	for _, want := range []string{"start nextcloud-db", "snapshot old", "dumpall", "stop nextcloud-db", "empty " + convVol, "drop-empty", "load", "snapshot new"} {
+		if !strings.Contains(order, want) {
+			t.Fatalf("the conversion never did %q: %s", want, order)
+		}
+	}
+	if strings.Index(order, "dumpall") > strings.Index(order, "empty") {
+		t.Fatalf("the volume was emptied before the dump was taken: %s", order)
+	}
+	if !strings.Contains(strings.Join(c.list(), ","), "up -d --no-deps db") {
+		t.Fatalf("the new engine was never started alone: %v", c.list())
+	}
+	cc := LoadAppConfig(dir).ConversionCopy
+	if cc == nil || cc.Volume != convVol || cc.From != 16 || cc.To != 18 {
+		t.Fatalf("the kept copy is not recorded: %+v", cc)
+	}
+	if fc.nCopies() != 1 || fc.copies[cc.Copy] != convOldData {
+		t.Fatalf("the OLD datadir's copy must be kept after success (B5); copies=%v", fc.copies)
+	}
+	if _, err := os.Stat(filepath.Join(dir, preUpdateConvertDir)); !os.IsNotExist(err) {
+		t.Fatal("the conversion's dump must be removed after success")
+	}
+}
+
+// TestConvert_MajorWithoutMarkIsRefused — B1's defence in depth: a step moving PostgreSQL across a major
+// with no engine_conversion mark is refused BEFORE anything moves, with the household sentence.
+//
+// COMPANION RED-PROOF (REPORT.md): at v0.272.0 there is no such check — the update pins 18 and runs;
+// reproduced by making planEngineConversion return nil, nil: this test fails at "the preflight let it through".
+func TestConvert_MajorWithoutMarkIsRefused(t *testing.T) {
+	m, dir, _, c, fc, pg := convManager(t, "")
+	ref := m.UpdatePreflight("nextcloud")
+	if ref == nil || ref.Reason != "engine_no_test" {
+		t.Fatalf("the preflight let it through: %+v", ref)
+	}
+	if want := "Ez a lépés a(z) Docmost adatbázisának fő verzióját váltaná, de nincs róla próba. Nem változott semmi."; ref.Error() != want {
+		t.Fatalf("sentence %q, want %q", ref.Error(), want)
+	}
+	if err := m.StartGuardedUpdate("nextcloud"); err == nil {
+		t.Fatal("StartGuardedUpdate accepted a PostgreSQL major move with no test")
+	}
+	if dbPin(t, dir) != "postgres:16-alpine" || fc.vol(convVol) != convOldData || len(c.list()) != 0 || len(pg.calls) != 0 {
+		t.Fatalf("something moved: pin=%s vol=%s compose=%v pg=%v", dbPin(t, dir), fc.vol(convVol), c.list(), pg.calls)
+	}
+}
+
+// TestConvert_JobRefusesAMarkThatDoesNotMatch — the job itself (not only the preflight) refuses a mark
+// naming a different major; nothing is pulled.
+func TestConvert_JobRefusesAMarkThatDoesNotMatch(t *testing.T) {
+	m, dir, _, c, _, _ := convManager(t, `{"service": "db", "engine": "postgres", "from": 16, "to": 17}`)
+	m.runGuardedUpdate(context.Background(), "nextcloud")
+	st, _ := m.GetStack("nextcloud")
+	if st.UpdatePhase != UpdatePhaseFailed || st.UpdateErrorKey != "err.stacks.update_engine_no_test" {
+		t.Fatalf("phase %q key %q, want failed with the no-test sentence", st.UpdatePhase, st.UpdateErrorKey)
+	}
+	if dbPin(t, dir) != "postgres:16-alpine" {
+		t.Fatal("the pin moved")
+	}
+	for _, call := range c.list() {
+		if strings.HasPrefix(call, "pull") {
+			t.Fatal("pulled before refusing")
+		}
+	}
+}
+
+func assertUndoneOnOld(t *testing.T, m *Manager, dir string, fc *fakeCopier) {
+	t.Helper()
+	st := waitUpdateDone(t, m, "nextcloud")
+	if st.UpdatePhase != UpdatePhaseUndone {
+		t.Fatalf("ended %q (%s), want undone", st.UpdatePhase, st.UpdateError)
+	}
+	if got := fc.vol(convVol); got != convOldData {
+		t.Fatalf("the database volume holds %q after the undo, want the OLD datadir back", got)
+	}
+	if got := dbPin(t, dir); got != "postgres:16-alpine" {
+		t.Fatalf("pin %q after the undo, want 16", got)
+	}
+	if LoadAppConfig(dir).ConversionCopy != nil {
+		t.Fatal("an undone conversion must not record a kept copy")
+	}
+}
+
+// TestConvert_CountsDifferAreUndone — the check after the load differs → undo, the old datadir back.
+func TestConvert_CountsDifferAreUndone(t *testing.T) {
+	m, dir, _, _, fc, pg := convManager(t, goodMark)
+	pg.after = []string{"db:docmost owner=docmost", "role:docmost super=true", "ext:docmost:pg_trgm", "rows:docmost:public.users=0"}
+	if err := m.StartGuardedUpdate("nextcloud"); err != nil {
+		t.Fatal(err)
+	}
+	assertUndoneOnOld(t, m, dir, fc)
+}
+
+// TestConvert_LoadFailureIsUndone — case (a) of Part D: the load errors → undo.
+func TestConvert_LoadFailureIsUndone(t *testing.T) {
+	m, dir, _, _, fc, pg := convManager(t, goodMark)
+	pg.loadErr = errors.New("ERROR: relation already exists")
+	if err := m.StartGuardedUpdate("nextcloud"); err != nil {
+		t.Fatal(err)
+	}
+	assertUndoneOnOld(t, m, dir, fc)
+}
+
+// TestConvert_HealthFailureIsUndone — case (b): converted fine, the app unhealthy on the new engine → undo.
+func TestConvert_HealthFailureIsUndone(t *testing.T) {
+	m, dir, _, _, fc, pg := convManager(t, goodMark)
+	m.updateHealthFn = func(context.Context, string, time.Duration) (bool, string) { return false, "new version unhealthy" }
+	if err := m.StartGuardedUpdate("nextcloud"); err != nil {
+		t.Fatal(err)
+	}
+	assertUndoneOnOld(t, m, dir, fc)
+	if !pg.called("load") {
+		t.Fatal("the fixture never reached the load — this test must fail AFTER the conversion")
+	}
+}
+
+// TestConvert_WrongVersionIsUndone — the new datadir's PG_VERSION is not the mark's `to` → undo.
+func TestConvert_WrongVersionIsUndone(t *testing.T) {
+	m, dir, _, _, fc, pg := convManager(t, goodMark)
+	pg.version = "17"
+	if err := m.StartGuardedUpdate("nextcloud"); err != nil {
+		t.Fatal(err)
+	}
+	assertUndoneOnOld(t, m, dir, fc)
+}
+
+// TestConvert_CutOffDumpNeverEmptiesTheVolume — a dump without its completion line stops the conversion
+// BEFORE the volume is touched.
+func TestConvert_CutOffDumpNeverEmptiesTheVolume(t *testing.T) {
+	m, dir, _, _, fc, pg := convManager(t, goodMark)
+	pg.dumpCut = true
+	if err := m.StartGuardedUpdate("nextcloud"); err != nil {
+		t.Fatal(err)
+	}
+	assertUndoneOnOld(t, m, dir, fc)
+	if pg.called("empty") {
+		t.Fatal("the volume was emptied after a cut-off dump")
+	}
+}
+
+// TestConvert_NeverEmptiedWithoutMarker — rule 3 of the brief: the DB volume is never emptied before the
+// undo copy's finished-marker is validated. The copy is made WITHOUT its marker; the volume must keep the
+// old datadir and the app must be held (the undo cannot use the copy either).
+//
+// COMPANION RED-PROOF (REPORT.md): drop the Complete() check before Empty in convertEngine AND the marker
+// test inside Empty — this test fails at "the volume was emptied".
+func TestConvert_NeverEmptiedWithoutMarker(t *testing.T) {
+	m, _, g, _, fc, pg := convManager(t, goodMark)
+	fc.cutOff = true
+	if err := m.StartGuardedUpdate("nextcloud"); err != nil {
+		t.Fatal(err)
+	}
+	st := waitUpdateDone(t, m, "nextcloud")
+	if fc.vol(convVol) != convOldData {
+		t.Fatalf("the volume was emptied (holds %q) without a validated copy", fc.vol(convVol))
+	}
+	if pg.called("empty") {
+		t.Fatal("Empty was called although the copy had no marker")
+	}
+	if held, _ := g.HoldFor("nextcloud"); !held || st.UpdatePhase != UpdatePhaseFailed {
+		t.Fatalf("held=%v phase=%q — with no usable copy the app must be HELD (data untouched)", held, st.UpdatePhase)
+	}
+}
+
+// TestConvert_NoSpaceIsRefusedWithNothingMoved — B6/A5: too little room beside the stack dir → refused
+// before the pin, the pull or the copy, with the household sentence naming both sizes.
+func TestConvert_NoSpaceIsRefusedWithNothingMoved(t *testing.T) {
+	m, dir, _, c, fc, pg := convManager(t, goodMark)
+	m.convertFreeFn = func(string) (int64, bool) { return 1 << 30, true }
+	m.runGuardedUpdate(context.Background(), "nextcloud")
+	st, _ := m.GetStack("nextcloud")
+	if st.UpdatePhase != UpdatePhaseFailed || st.UpdateErrorKey != "err.stacks.update_convert_space" {
+		t.Fatalf("phase %q key %q", st.UpdatePhase, st.UpdateErrorKey)
+	}
+	if !strings.HasPrefix(st.UpdateError, "A(z) Docmost adatbázisának átalakításához ") || !strings.HasSuffix(st.UpdateError, "Nem változott semmi.") {
+		t.Fatalf("sentence %q", st.UpdateError)
+	}
+	if dbPin(t, dir) != "postgres:16-alpine" || fc.nCopies() != 0 || len(pg.calls) != 0 {
+		t.Fatalf("something moved: pin=%s copies=%d pg=%v", dbPin(t, dir), fc.nCopies(), pg.calls)
+	}
+	for _, call := range c.list() {
+		if strings.HasPrefix(call, "pull") || strings.HasPrefix(call, "stop") {
+			t.Fatalf("compose %q ran before the refusal", call)
+		}
+	}
+}
+
+// TestConvert_RestartDuringConvertingIsUndone — B4: the controller dies after the volume was emptied; the
+// journal says `converting`; the next start UNDOES it — the old datadir back, the old pin, `undone`.
+func TestConvert_RestartDuringConvertingIsUndone(t *testing.T) {
+	m, dir, _, _, fc, pg := convManager(t, goodMark)
+	killed := make(chan struct{})
+	var once sync.Once
+	pg.onConvert = func(step string) {
+		if strings.HasPrefix(step, "empty") {
+			once.Do(func() { close(killed) })
+			select {} // the process "dies" here: this goroutine never returns
+		}
+	}
+	go m.runGuardedUpdate(context.Background(), "nextcloud")
+	select {
+	case <-killed:
+	case <-time.After(5 * time.Second):
+		t.Fatal("never reached the empty step")
+	}
+	time.Sleep(50 * time.Millisecond) // let Empty's own write land
+	fc.setVol(convVol, "")            // what a real kill leaves: the volume emptied
+
+	// a NEW manager process reading the same disk
+	m2, _, _, _, _, _ := convManager(t, goodMark)
+	m2.cfg, m2.stacks, m2.undoCopier = m.cfg, map[string]*Stack{"nextcloud": {Name: "nextcloud", Deployed: true, ComposePath: filepath.Join(dir, "docker-compose.yml"), AppConfig: LoadAppConfig(dir)}}, fc
+	m2.pgConv = &fakePG{fc: fc}
+	m2.updateGuards = m.updateGuards
+	j := m2.readUpdateJournal().Updates["nextcloud"]
+	if j.Phase != UpdatePhaseConverting || !j.ConvertTouched || !j.Copied {
+		t.Fatalf("journal phase=%q touched=%v copied=%v — the restart must find `converting` with the copies", j.Phase, j.ConvertTouched, j.Copied)
+	}
+	if got := m2.RecoverUpdates(); len(got) != 1 {
+		t.Fatalf("RecoverUpdates resumed %v", got)
+	}
+	m2.ResumeInterruptedUpdates(context.Background())
+	assertUndoneOnOld(t, m2, dir, fc)
+}
+
+// TestConvert_ReleaseAfterAProvenBackup — B5: the kept copy stays while no backup is proven after the
+// conversion, and goes (with its record) once one is.
+func TestConvert_ReleaseAfterAProvenBackup(t *testing.T) {
+	m, dir, g, _, fc, _ := convManager(t, goodMark)
+	if err := m.StartGuardedUpdate("nextcloud"); err != nil {
+		t.Fatal(err)
+	}
+	if st := waitUpdateDone(t, m, "nextcloud"); st.UpdatePhase != UpdatePhaseDone {
+		t.Fatalf("setup: %q", st.UpdatePhase)
+	}
+	if err := m.ScanStacks(); err == nil {
+		_ = err
+	}
+	m.mu.Lock()
+	m.stacks["nextcloud"].AppConfig = LoadAppConfig(dir)
+	m.mu.Unlock()
+	if got := m.ReleaseConversionCopies(context.Background()); len(got) != 0 || fc.nCopies() != 1 {
+		t.Fatalf("released %v with only a pre-conversion backup; copies=%d", got, fc.nCopies())
+	}
+	g.mu.Lock()
+	g.points = []UpdateRestorePoint{{Tier: UpdateTierLocal, ProvenAt: slice4T0.Add(time.Hour)}}
+	g.mu.Unlock()
+	if got := m.ReleaseConversionCopies(context.Background()); len(got) != 1 || fc.nCopies() != 0 {
+		t.Fatalf("released %v after a proven backup; copies=%d", got, fc.nCopies())
+	}
+	if LoadAppConfig(dir).ConversionCopy != nil {
+		t.Fatal("the record must go with the copy")
+	}
+}
+
+func TestConvert_PostgresMajorFromTags(t *testing.T) {
+	for ref, want := range map[string]int{
+		"postgres:16-alpine": 16, "postgres:17.2": 17, "postgres:18-alpine@sha256:ab": 18,
+		"postgis/postgis:16-3.5-alpine": 16, "pgvector/pgvector:pg16": 16,
+		"ghcr.io/immich-app/postgres:16-vectorchord0.4.3-pgvectors0.2.0": 16,
+	} {
+		if got, ok := postgresMajor(ref); !ok || got != want {
+			t.Errorf("%s → %d %v, want %d", ref, got, ok, want)
+		}
+	}
+	if _, ok := postgresMajor("postgres"); ok {
+		t.Error("an untagged ref has no major")
+	}
+	// within a major is not a conversion; a non-postgres move is not either
+	if c, err := planEngineConversion(map[string]string{"db": "postgres:16-alpine"}, map[string]string{"db": "postgres:16.4-alpine"}, nil); c != nil || err != nil {
+		t.Errorf("16 → 16.4 must be neither a conversion nor a refusal: %v %v", c, err)
+	}
+	if c, err := planEngineConversion(map[string]string{"r": "redis:7"}, map[string]string{"r": "redis:8"}, nil); c != nil || err != nil {
+		t.Errorf("redis is not postgres: %v %v", c, err)
+	}
+}
+
+// TestConvert_PostgresFamilyMatchesTheBackupSide — isPostgresImage must know every family the backup
+// side dumps as Postgres (appbackup.dbTypeForImage, R-484); read from its source, the only way across the
+// package boundary without exporting it.
+func TestConvert_PostgresFamilyMatchesTheBackupSide(t *testing.T) {
+	src, err := os.ReadFile("../appbackup/dbservices.go")
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, fam := range []string{"postgres", "postgis", "pgvector", "timescaledb"} {
+		if !strings.Contains(string(src), `strings.Contains(img, "`+fam+`")`) {
+			t.Fatalf("the backup side no longer names %q — re-read it and update isPostgresImage", fam)
+		}
+		if !isPostgresImage("x/" + fam + ":1") {
+			t.Errorf("isPostgresImage misses %q", fam)
+		}
+	}
+}
+
+func TestConvert_FilterSkipsOnlyExactLines(t *testing.T) {
+	in := "CREATE ROLE docmost;\nCREATE ROLE docmost_ro;\nALTER ROLE docmost WITH PASSWORD 'x';\nCREATE ROLE docmost;  \n"
+	var out bytes.Buffer
+	if err := filterLines(strings.NewReader(in), &out, map[string]bool{"CREATE ROLE docmost;": true}); err != nil {
+		t.Fatal(err)
+	}
+	want := "CREATE ROLE docmost_ro;\nALTER ROLE docmost WITH PASSWORD 'x';\nCREATE ROLE docmost;  \n"
+	if out.String() != want {
+		t.Fatalf("got %q", out.String())
+	}
+}
+
+func TestConvert_ValidateDumpAll(t *testing.T) {
+	d := t.TempDir()
+	good, cut := filepath.Join(d, "g"), filepath.Join(d, "c")
+	mustWrite(t, good, strings.Repeat("x", 9000)+"\n-- "+dumpAllCompleteLine+"\n")
+	mustWrite(t, cut, strings.Repeat("x", 9000)+"\n-- PostgreSQL database dump complete\n")
+	if err := validateDumpAll(good); err != nil {
+		t.Fatal(err)
+	}
+	if validateDumpAll(cut) == nil {
+		t.Fatal("a dump ending with a PER-DATABASE completion line is not a whole cluster dump")
+	}
+}
+
+// TestConvert_MarkDoesNotChangeOldLadderPrints — the new field is omitempty, so R-680's failed-step
+// records (tied to LadderPrint) survive the upgrade to v0.273.0.
+func TestConvert_MarkDoesNotChangeOldLadderPrints(t *testing.T) {
+	b, _ := json.Marshal([]LadderEntry{{From: map[string]string{"a": "x:1"}, To: map[string]string{"a": "x:2"}, Verdict: "proven"}})
+	if strings.Contains(string(b), "engine_conversion") {
+		t.Fatalf("an entry without the mark prints it: %s", b)
+	}
+}

controller/internal/stacks/undo.go

@@ -88,7 +88,9 @@ const appliedMetaDir = "applied-meta"
 
 // AppliedMetaFile is the pinned version's .felhom.yml record (R-669, v0.270.0: the recovery unit captures
 // it, so a restore brings back the PINNED version's health check, not the sync's newer one).
-func AppliedMetaFile(stackDir string) string { return filepath.Join(stackDir, appliedMetaDir, ".felhom.yml") }
+func AppliedMetaFile(stackDir string) string {
+	return filepath.Join(stackDir, appliedMetaDir, ".felhom.yml")
+}
 
 // RecordRestoredAppliedMeta makes the .felhom.yml a restore just wrote the pinned version's record
 // (R-669, v0.270.0). MEASURED 2026-09-24 (A2): after a failed step ended in a restore, applied-meta still
@@ -555,6 +557,7 @@ func (m *Manager) tryUndo(ctx context.Context, name, dir, why string, entry *upd
 	m.recordUpdateUndone(name, dir, &UpdateUndone{To: entry.NewPin, At: start.UTC().Format(time.RFC3339), Why: why})
 	m.recordFailedStep(name, dir, entry.NewPin, "undone") // R-680
 	m.removePreUpdateCopies(dir)
+	_ = os.RemoveAll(filepath.Join(dir, preUpdateConvertDir)) // v0.273.0: the conversion's dump, if any
 	_ = m.RefreshStatus()
 	m.clearJournal(name)
 	if err := m.ScanStacks(); err != nil { // R-678: the page reads the pin the undo put back

controller/internal/stacks/update.go

@@ -79,6 +79,8 @@ var updatePhaseLabels = map[string]string{
 	UpdatePhaseCopying: "Az adatok másolása a frissítés előtt…",
 	UpdatePhaseUndoing: "Visszaállítás az előző változatra…",
 	UpdatePhaseUndone:  "Visszaállítva az előző változatra",
+	// v0.273.0 — born as a bundle key (update.phase.converting).
+	UpdatePhaseConverting: "Adatbázis átalakítása",
 }
 
 // UpdatePhaseLabel returns the customer label for a phase, "" for an unknown one.
@@ -444,6 +446,14 @@ func (m *Manager) UpdatePreflight(name string) *UpdateRefusal {
 		}
 		m.logger.Printf("[WARN] [stacks] update %s: no backup can be taken now (%s) — an existing copy must carry the update", name, why)
 	}
+	// v0.273.0 (B1) — a PostgreSQL major move without the test's mark is refused before anything moves.
+	if st.AppConfig != nil && len(st.AppConfig.PinnedImages) > 0 {
+		if step, err := nextLadderStep(filepath.Dir(m.CatalogTemplatePath(name, "docker-compose.yml")), st.AppConfig.PinnedImages); err == nil {
+			if _, _, key, cerr := m.conversionFor(st, st.AppConfig.PinnedImages, step); cerr != nil && key == "err.stacks.update_engine_no_test" {
+				return m.refuseUpdateErr(name, "engine_no_test", util.MsgError(key, appLabel(st)), cerr.Error())
+			}
+		}
+	}
 	if ref := m.updateMemoryRefusal(name, st); ref != nil {
 		return ref
 	}
@@ -774,6 +784,21 @@ func (m *Manager) runGuardedUpdate(ctx context.Context, name string) {
 		}
 		stepSrc, stepMeta = step.Source, step.Meta
 		m.logger.Printf("[INFO] [stacks] update %s: ladder — %s", name, step.Why)
+		// v0.273.0 — A POSTGRESQL MAJOR MOVES ONLY WITH THE TEST'S MARK (`09` §6.4 part 10, B1). Decided
+		// here, before anything moves; the preflight asked the same question (conversionRefusal).
+		conv, vol, key, cerr := m.conversionFor(st, cfg.PinnedImages, step)
+		if cerr != nil {
+			if key == "" {
+				fail("update.error.pin_failed", "engine conversion: "+cerr.Error())
+			} else {
+				fail(key, "engine conversion: "+cerr.Error(), appLabel(st))
+			}
+			return
+		}
+		if conv != nil {
+			entry.Convert, entry.ConvertVolume = conv, vol
+			m.logger.Printf("[INFO] [stacks] update %s: this step CONVERTS %s PostgreSQL %d → %d (the ladder entry's mark); database volume %s", name, conv.Service, conv.From, conv.To, vol)
+		}
 	}
 	// R-665 (v0.269.0): the new version is judged by ITS OWN .felhom.yml, journaled so a resumed verify
 	// uses it too — never by the stack dir's, which a restore rewrites with the unit's older file until
@@ -831,6 +856,24 @@ func (m *Manager) runGuardedUpdate(ctx context.Context, name string) {
 		}
 		return
 	}
+	if entry.Convert != nil {
+		covered := false
+		for _, v := range undoVols {
+			covered = covered || v == entry.ConvertVolume
+		}
+		if !covered {
+			fail("update.error.pin_failed", fmt.Sprintf("engine conversion: the database volume %s is not in the undo copy %v — refusing before anything moves", entry.ConvertVolume, undoVols))
+			return
+		}
+		if err := m.planConversionSpace(name, dir, entry.ConvertVolume); err != nil {
+			if se, ok := err.(*convertSpaceError); ok {
+				fail("err.stacks.update_convert_space", "engine conversion: "+err.Error(), appLabel(st), humanBytes(se.need), humanBytes(se.free))
+			} else {
+				fail("err.stacks.update_undo_copy_failed", "engine conversion space: "+err.Error())
+			}
+			return
+		}
+	}
 
 	// PINNING — the previous definition is copied aside and journaled BEFORE the pin moves, so a crash
 	// at any later instant can put it back (Scenario G).
@@ -914,6 +957,18 @@ func (m *Manager) runGuardedUpdate(ctx context.Context, name string) {
 		return
 	}
 
+	// CONVERTING (v0.273.0) — only on a step the ladder marks. Any failure is undone like a failed health
+	// check: every volume back from its copy (the old datadir included), old pin, old definition.
+	if entry.Convert != nil {
+		if !m.enterUpdatePhase(name, &entry, UpdatePhaseConverting) {
+			m.failAndHold(ctx, name, dir, env, rp, "journal write failed before converting", &entry)
+			return
+		}
+		if err := m.convertEngine(ctx, name, dir, env, &entry); err != nil {
+			m.failAndHold(ctx, name, dir, env, rp, "conversion failed: "+err.Error(), &entry)
+			return
+		}
+	}
 	if !m.enterUpdatePhase(name, &entry, UpdatePhaseStarting) {
 		m.failAndHold(ctx, name, dir, env, rp, "journal write failed before up", &entry)
 		return
@@ -942,7 +997,27 @@ func (m *Manager) verifyAndConclude(ctx context.Context, name, dir string, env [
 	m.logger.Printf("[INFO] [stacks] update %s: healthy after %s (%s)", name, m.now().Sub(waitStart).Round(time.Second), detail)
 	m.recordInstalledImages(name, dir, env)
 	_ = m.RefreshStatus()
-	m.removeUndoCopies(name, entry.UndoCopies)
+	if entry.Convert != nil {
+		// B5 (v0.273.0): the OLD datadir's copy stays until a backup of the converted app is proven
+		// (ReleaseConversionCopies); every other copy goes as usual.
+		var keep *undoCopy
+		var rest []undoCopy
+		for i, c := range entry.UndoCopies {
+			if c.Volume == entry.ConvertVolume {
+				keep = &entry.UndoCopies[i]
+				continue
+			}
+			rest = append(rest, c)
+		}
+		m.removeUndoCopies(name, rest)
+		if keep != nil {
+			m.recordConversionCopy(name, dir, &ConversionCopy{Volume: keep.Volume, Copy: keep.Copy, At: m.now().UTC().Format(time.RFC3339), From: entry.Convert.From, To: entry.Convert.To})
+			m.logger.Printf("[INFO] [stacks] update %s: KEEPING the pre-conversion datadir copy %s (PostgreSQL %d) until a backup of the converted app is proven", name, keep.Copy, entry.Convert.From)
+		}
+		_ = os.RemoveAll(filepath.Join(dir, preUpdateConvertDir))
+	} else {
+		m.removeUndoCopies(name, entry.UndoCopies)
+	}
 	m.recordUpdateUndone(name, dir, nil) // a successful update ends the "undone" note
 	m.clearFailedStep(name, dir)         // R-680: and the failed-step record
 	m.clearJournal(name)
@@ -1203,6 +1278,11 @@ type updateJournalEntry struct {
 	// NewMeta (v0.269.0, R-665/R-664) is the NEW version's own .felhom.yml — the catalog's, or the
 	// ladder step's — used for the verify, so a resumed verify judges by the same file.
 	NewMeta string `json:"new_meta,omitempty"`
+	// v0.273.0 (pgconvert.go): the step's engine conversion, the DB volume it rebuilds, and whether that
+	// volume has been touched (emptied) — journaled so a restart during `converting` runs the undo.
+	Convert        *EngineConversion `json:"convert,omitempty"`
+	ConvertVolume  string            `json:"convert_volume,omitempty"`
+	ConvertTouched bool              `json:"convert_touched,omitempty"`
 }
 
 type updateJournal struct {
@@ -1343,7 +1423,9 @@ func (m *Manager) RecoverUpdates() []string {
 			}
 			m.clearJournal(name)
 			m.finishUpdateKey(name, UpdatePhaseFailed, "update.error.interrupted", "")
-		case UpdatePhaseUndoing:
+		case UpdatePhaseUndoing, UpdatePhaseConverting:
+			// v0.273.0: a restart during `converting` is UNDONE the same way — the database volume may be
+			// emptied or half-loaded, and only the copy is known-good (B4). Never "done".
 			// v0.263.0: a power cut DURING the undo. Resumed like `starting` — the undo runs again from
 			// the copies (still there: they are removed only after the undo succeeded) and then probes.
 			// Never "done": what ran last was a failed new version.
@@ -1399,9 +1481,13 @@ func (m *Manager) ResumeInterruptedUpdates(ctx context.Context) int {
 		dir := filepath.Dir(st.ComposePath)
 		go func(name, dir string, e updateJournalEntry, rp UpdateRestorePoint) {
 			env := m.stackEnv(dir)
-			if e.Phase == UpdatePhaseUndoing {
-				m.logger.Printf("[INFO] [stacks] update %s: resuming the UNDO after a controller restart", name)
-				m.failAndHold(ctx, name, dir, env, rp, "resumed after a restart during the undo", &e)
+			if e.Phase == UpdatePhaseUndoing || e.Phase == UpdatePhaseConverting {
+				why := "resumed after a restart during the undo"
+				if e.Phase == UpdatePhaseConverting {
+					why = "the controller restarted during the database conversion — undoing it"
+				}
+				m.logger.Printf("[INFO] [stacks] update %s: resuming the UNDO after a controller restart (was %s)", name, e.Phase)
+				m.failAndHold(ctx, name, dir, env, rp, why, &e)
 				return
 			}
 			m.logger.Printf("[INFO] [stacks] update %s: resuming after a controller restart — `up -d` then the health wait", name)