feat(state): C1 — saga_phase v2 migration + resumability CRUD

PROGRESS.md

@@ -19,7 +19,7 @@ and verification. **Merge model:** PR-per-chunk, auto-merge on green CI.
 ## Frontier
 
 ### M2-remainder (core saga)
-- [ ] C1  state v2 `saga_phase` table + CRUD — TODO  *(kickoff)*
+- [x] C1  state v2 `saga_phase` table + CRUD — REVIEW (PR `nightly/C1`)
 - [ ] C2  docker `ContainerInspect`/`ContainerLogs` + mock — TODO  *(kickoff)*
 - [ ] C3a config `healthcheck:`/`hooks:` structs — TODO  *(kickoff)*
 - [ ] C3b `internal/health` thin poller — BLOCKED (C2, C3a)

internal/state/migrations.go

@@ -13,6 +13,7 @@ type migration struct {
 // or remove a released one.
 var migrations = []migration{
 	{version: 1, stmt: schemaV1},
+	{version: 2, stmt: schemaV2},
 }
 
 // schemaV1 is the initial ledger (spec 08 §Tables). Every mutable row is scoped
@@ -94,6 +95,25 @@ CREATE TABLE IF NOT EXISTS schema_version (
 );
 `
 
+// schemaV2 (spec 08 §saga_phase) adds the up-saga resumability table: one row per
+// (workspace, scope, phase) recording its status + an input fingerprint, so a
+// re-run skips satisfied phases and a crash mid-saga resumes deterministically.
+const schemaV2 = `
+CREATE TABLE IF NOT EXISTS saga_phase (
+    ctx          TEXT NOT NULL,
+    workspace    TEXT NOT NULL,
+    scope        TEXT NOT NULL DEFAULT '',   -- '' = workspace-wide, else project name
+    phase        TEXT NOT NULL,              -- preflight|clone|network|shared|provision|secrets|generate|compose-up|hooks|...
+    status       TEXT NOT NULL,              -- pending|started|satisfied|failed
+    fingerprint  TEXT NOT NULL DEFAULT '',   -- SHA-256 of the phase's inputs
+    started_at   TEXT,
+    satisfied_at TEXT,
+    error        TEXT,
+    PRIMARY KEY (ctx, workspace, scope, phase),
+    FOREIGN KEY (ctx) REFERENCES docker_context(name) ON DELETE CASCADE
+);
+`
+
 // migrate applies any pending migrations inside a transaction per step, backing
 // up the DB file before the first mutating step. Forward-only.
 func (db *DB) migrate() error {

internal/state/saga.go

@@ -0,0 +1,126 @@
+package state
+
+import (
+	"database/sql"
+	"fmt"
+)
+
+// Saga phase statuses.
+const (
+	PhasePending   = "pending"
+	PhaseStarted   = "started"
+	PhaseSatisfied = "satisfied"
+	PhaseFailed    = "failed"
+)
+
+// SagaPhase is one row of the saga_phase resumability table (spec 08/09). A phase
+// is keyed by (workspace, scope, phase); scope is "" for workspace-wide phases or
+// a project name for per-project ones.
+type SagaPhase struct {
+	Workspace   string
+	Scope       string
+	Phase       string
+	Status      string
+	Fingerprint string
+	StartedAt   string
+	SatisfiedAt string
+	Error       string
+}
+
+// StartPhase marks a phase started with the given input fingerprint (upsert).
+// Clears any prior satisfied/failed/error state for a fresh attempt. Hold the lock.
+func (db *DB) StartPhase(workspace, scope, phase, fingerprint string) error {
+	_, err := db.Exec(`
+		INSERT INTO saga_phase (ctx, workspace, scope, phase, status, fingerprint, started_at)
+		VALUES (?,?,?,?,?,?,datetime('now'))
+		ON CONFLICT(ctx, workspace, scope, phase)
+		DO UPDATE SET status=excluded.status, fingerprint=excluded.fingerprint,
+		             started_at=excluded.started_at, satisfied_at=NULL, error=NULL`,
+		db.Ctx, workspace, scope, phase, PhaseStarted, fingerprint)
+	if err != nil {
+		return fmt.Errorf("start phase %s/%s/%s: %w", workspace, scope, phase, err)
+	}
+	return nil
+}
+
+// SatisfyPhase marks a phase satisfied (completed). Hold the lock.
+func (db *DB) SatisfyPhase(workspace, scope, phase string) error {
+	_, err := db.Exec(`UPDATE saga_phase SET status=?, satisfied_at=datetime('now'), error=NULL
+		WHERE ctx=? AND workspace=? AND scope=? AND phase=?`,
+		PhaseSatisfied, db.Ctx, workspace, scope, phase)
+	if err != nil {
+		return fmt.Errorf("satisfy phase %s/%s/%s: %w", workspace, scope, phase, err)
+	}
+	return nil
+}
+
+// FailPhase marks a phase failed with an error message. Hold the lock.
+func (db *DB) FailPhase(workspace, scope, phase, errMsg string) error {
+	_, err := db.Exec(`UPDATE saga_phase SET status=?, error=?
+		WHERE ctx=? AND workspace=? AND scope=? AND phase=?`,
+		PhaseFailed, errMsg, db.Ctx, workspace, scope, phase)
+	if err != nil {
+		return fmt.Errorf("fail phase %s/%s/%s: %w", workspace, scope, phase, err)
+	}
+	return nil
+}
+
+// GetPhase returns a phase row, ok=false if it has never been recorded.
+func (db *DB) GetPhase(workspace, scope, phase string) (SagaPhase, bool, error) {
+	var p SagaPhase
+	var started, satisfied, errMsg sql.NullString
+	err := db.QueryRow(`SELECT workspace, scope, phase, status, fingerprint, started_at, satisfied_at, error
+		FROM saga_phase WHERE ctx=? AND workspace=? AND scope=? AND phase=?`,
+		db.Ctx, workspace, scope, phase).
+		Scan(&p.Workspace, &p.Scope, &p.Phase, &p.Status, &p.Fingerprint, &started, &satisfied, &errMsg)
+	if err == sql.ErrNoRows {
+		return SagaPhase{}, false, nil
+	}
+	if err != nil {
+		return SagaPhase{}, false, fmt.Errorf("get phase: %w", err)
+	}
+	p.StartedAt, p.SatisfiedAt, p.Error = started.String, satisfied.String, errMsg.String
+	return p, true, nil
+}
+
+// PhaseSatisfied reports whether a phase is already satisfied for the SAME input
+// fingerprint — the skip check a re-run uses to avoid redoing unchanged work. A
+// changed fingerprint (config/template/param edit) re-arms the phase.
+func (db *DB) PhaseSatisfied(workspace, scope, phase, fingerprint string) (bool, error) {
+	p, ok, err := db.GetPhase(workspace, scope, phase)
+	if err != nil || !ok {
+		return false, err
+	}
+	return p.Status == PhaseSatisfied && p.Fingerprint == fingerprint, nil
+}
+
+// PhasesFor returns all recorded phases for a workspace, ordered, for `status`.
+func (db *DB) PhasesFor(workspace string) ([]SagaPhase, error) {
+	rows, err := db.Query(`SELECT workspace, scope, phase, status, fingerprint, started_at, satisfied_at, error
+		FROM saga_phase WHERE ctx=? AND workspace=? ORDER BY scope, phase`, db.Ctx, workspace)
+	if err != nil {
+		return nil, fmt.Errorf("phases for %s: %w", workspace, err)
+	}
+	defer rows.Close()
+	var out []SagaPhase
+	for rows.Next() {
+		var p SagaPhase
+		var started, satisfied, errMsg sql.NullString
+		if err := rows.Scan(&p.Workspace, &p.Scope, &p.Phase, &p.Status, &p.Fingerprint, &started, &satisfied, &errMsg); err != nil {
+			return nil, err
+		}
+		p.StartedAt, p.SatisfiedAt, p.Error = started.String, satisfied.String, errMsg.String
+		out = append(out, p)
+	}
+	return out, rows.Err()
+}
+
+// ClearPhase removes a phase row (compensation / forced re-run). Hold the lock.
+func (db *DB) ClearPhase(workspace, scope, phase string) error {
+	_, err := db.Exec(`DELETE FROM saga_phase WHERE ctx=? AND workspace=? AND scope=? AND phase=?`,
+		db.Ctx, workspace, scope, phase)
+	if err != nil {
+		return fmt.Errorf("clear phase %s/%s/%s: %w", workspace, scope, phase, err)
+	}
+	return nil
+}

internal/state/saga_test.go

@@ -0,0 +1,90 @@
+package state
+
+import "testing"
+
+func TestSagaPhaseLifecycle(t *testing.T) {
+	db := openTestDB(t)
+
+	// Unrecorded phase: not satisfied, GetPhase ok=false.
+	if ok, _ := db.PhaseSatisfied("acme", "", "network", "fp1"); ok {
+		t.Fatal("unrecorded phase should not be satisfied")
+	}
+	if _, ok, _ := db.GetPhase("acme", "", "network"); ok {
+		t.Fatal("unrecorded phase should report ok=false")
+	}
+
+	// Start → satisfied for fingerprint fp1.
+	if err := db.StartPhase("acme", "", "network", "fp1"); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.SatisfyPhase("acme", "", "network"); err != nil {
+		t.Fatal(err)
+	}
+	if ok, _ := db.PhaseSatisfied("acme", "", "network", "fp1"); !ok {
+		t.Error("phase should be satisfied for the same fingerprint")
+	}
+	// A changed fingerprint re-arms the phase (skip check fails).
+	if ok, _ := db.PhaseSatisfied("acme", "", "network", "fp2"); ok {
+		t.Error("a changed fingerprint must NOT count as satisfied")
+	}
+
+	p, ok, err := db.GetPhase("acme", "", "network")
+	if err != nil || !ok {
+		t.Fatalf("get: ok=%v err=%v", ok, err)
+	}
+	if p.Status != PhaseSatisfied || p.SatisfiedAt == "" || p.StartedAt == "" {
+		t.Errorf("phase row = %+v", p)
+	}
+}
+
+func TestSagaPhaseRestartClearsState(t *testing.T) {
+	db := openTestDB(t)
+	_ = db.StartPhase("acme", "api", "provision", "fpA")
+	_ = db.FailPhase("acme", "api", "provision", "boom")
+
+	p, _, _ := db.GetPhase("acme", "api", "provision")
+	if p.Status != PhaseFailed || p.Error != "boom" {
+		t.Fatalf("expected failed with error, got %+v", p)
+	}
+	// Re-start clears the prior error + satisfied_at.
+	if err := db.StartPhase("acme", "api", "provision", "fpB"); err != nil {
+		t.Fatal(err)
+	}
+	p, _, _ = db.GetPhase("acme", "api", "provision")
+	if p.Status != PhaseStarted || p.Error != "" || p.Fingerprint != "fpB" {
+		t.Errorf("restart did not reset state: %+v", p)
+	}
+}
+
+func TestSagaPhasesForAndClear(t *testing.T) {
+	db := openTestDB(t)
+	_ = db.StartPhase("acme", "", "network", "f")
+	_ = db.SatisfyPhase("acme", "", "network")
+	_ = db.StartPhase("acme", "api", "compose-up", "f")
+
+	got, err := db.PhasesFor("acme")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(got) != 2 {
+		t.Fatalf("phases = %d, want 2", len(got))
+	}
+
+	if err := db.ClearPhase("acme", "", "network"); err != nil {
+		t.Fatal(err)
+	}
+	if got, _ := db.PhasesFor("acme"); len(got) != 1 {
+		t.Fatalf("phases after clear = %d, want 1", len(got))
+	}
+}
+
+func TestSchemaVersionIsTwo(t *testing.T) {
+	db := openTestDB(t)
+	v, err := db.SchemaVersion()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if v != 2 {
+		t.Errorf("schema version = %d, want 2 (saga_phase migration applied)", v)
+	}
+}