//go:build linux

package hostexec

import (
	"context"
	"encoding/json"
	"errors"
	"os"
	"os/exec"
	"path/filepath"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/release"
	"golang.org/x/sys/unix"
)

type brokerDisk struct {
	dir         string
	writes      int
	failAt      int
	afterRename func(string, []byte)
}

func (d *brokerDisk) Read(n string) ([]byte, error) { return os.ReadFile(filepath.Join(d.dir, n)) }
func (d *brokerDisk) Write(n string, b []byte) error {
	d.writes++
	f, err := os.CreateTemp(d.dir, "pending-")
	if err != nil {
		return err
	}
	name := f.Name()
	defer os.Remove(name)
	if _, err = f.Write(b); err == nil {
		err = f.Sync()
	}
	err = errors.Join(err, f.Close())
	if err != nil {
		return err
	}
	if err = os.Rename(name, filepath.Join(d.dir, n)); err != nil {
		return err
	}
	if d.afterRename != nil {
		d.afterRename(n, b)
	}
	if d.writes == d.failAt {
		return os.ErrInvalid
	}
	parent, err := os.Open(d.dir)
	if err != nil {
		return err
	}
	return errors.Join(parent.Sync(), parent.Close())
}
func (d *brokerDisk) Remove(n string) error { return os.Remove(filepath.Join(d.dir, n)) }
func (d *brokerDisk) Names() ([]string, error) {
	es, e := os.ReadDir(d.dir)
	var out []string
	for _, v := range es {
		out = append(out, v.Name())
	}
	return out, e
}
func (d *brokerDisk) Lock(ctx context.Context, name string) (func(), error) {
	f, err := os.OpenFile(filepath.Join(d.dir, name), os.O_CREATE|os.O_RDWR, 0600)
	if err != nil {
		return nil, err
	}
	for {
		if err := ctx.Err(); err != nil {
			f.Close()
			return nil, err
		}
		err = unix.Flock(int(f.Fd()), unix.LOCK_EX|unix.LOCK_NB)
		if err == nil {
			return func() { _ = unix.Flock(int(f.Fd()), unix.LOCK_UN); _ = f.Close() }, nil
		}
		if !errors.Is(err, unix.EWOULDBLOCK) && !errors.Is(err, unix.EAGAIN) {
			f.Close()
			return nil, err
		}
		select {
		case <-ctx.Done():
			f.Close()
			return nil, ctx.Err()
		case <-time.After(time.Millisecond):
		}
	}
}
func (d *brokerDisk) Close() error { return nil }

func brokerFixture(t *testing.T) (*RootBroker, *brokerDisk, *delayedStore, *testClock, ExecutionRequest, ControllerProcess) {
	t.Helper()
	return brokerFixtureAt(t, t.TempDir(), t.TempDir())
}
func brokerFixtureAt(t *testing.T, local, remotePath string) (*RootBroker, *brokerDisk, *delayedStore, *testClock, ExecutionRequest, ControllerProcess) {
	t.Helper()
	d := &brokerDisk{dir: local}
	clock := &testClock{wall: time.Date(2026, 10, 13, 12, 0, 0, 0, time.UTC), boot: int64(100 * time.Second)}
	remote := &delayedStore{Store: release.NewDirStore(remotePath)}
	q := ExecutionRequest{Schema: 1, ID: "begin-1", Operation: "executionBegin", Host: "shop.example.com", Incarnation: "inc", Stamp: 3, Session: 2, Op: "move-1"}
	p := ControllerProcess{KernelBootID: "boot", PID: 42, Start: "start-42"}
	b, err := NewRootBroker(RootBrokerConfig{State: d, Remote: remote, Clock: clock, BootID: "boot", Identity: "broker-1",
		Authority: func(_ context.Context, got ExecutionRequest, peer ControllerProcess) (ExecutionAuthority, error) {
			if got.Host != q.Host || got.Incarnation != q.Incarnation || got.Stamp != q.Stamp || got.Session != q.Session || got.Op != q.Op || peer != p {
				return ExecutionAuthority{}, ErrInactive
			}
			return ExecutionAuthority{Box: "box", Enrollment: "en", Account: "bab", App: "jam", Installation: "installation-1", Migration: 1, ClaimRevision: "claim-1", RecordRevision: "record-1", DirectoryRevision: "directory-1", Kind: "landing", Phase: "checking"}, nil
		}, Alive: func(context.Context, ControllerProcess) error { return nil }, Cleanup: func(context.Context, ExecutionRecord) error { return nil }, Settle: func(context.Context, ExecutionRecord) error { return nil },
		Exclude: func(_ context.Context, fn func() error) error { return fn() },
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = b.Close(context.Background()) })
	return b, d, remote, clock, q, p
}

func TestPlan14ExecutionBrokerPeerAndShape(t *testing.T) {
	for _, mutation := range []string{"schema", "host", "incarnation", "stamp", "session", "operation", "peer-start", "peer-pid", "peer-boot"} {
		t.Run(mutation, func(t *testing.T) {
			b, d, remote, _, q, p := brokerFixture(t)
			switch mutation {
			case "schema":
				q.Schema++
			case "host":
				q.Host = "other.example.com"
			case "incarnation":
				q.Incarnation = "other"
			case "stamp":
				q.Stamp++
			case "session":
				q.Session++
			case "operation":
				q.Op = "other"
			case "peer-start":
				p.Start = "reused"
			case "peer-pid":
				p.PID++
			case "peer-boot":
				p.KernelBootID = "reboot"
			}
			if _, err := b.Execute(t.Context(), p, q); err == nil {
				t.Fatal("altered identity acquired participation")
			}
			if remote.calls != 0 || d.writes != 0 {
				t.Fatalf("refusal had effects: remote=%d writes=%d", remote.calls, d.writes)
			}
		})
	}
	b, d, remote, _, q, p := brokerFixture(t)
	first, err := b.Execute(t.Context(), p, q)
	if err != nil {
		t.Fatal(err)
	}
	writes := d.writes
	again, err := b.Execute(t.Context(), p, q)
	if err != nil || first != again {
		t.Fatalf("replay: %+v %+v %v", first, again, err)
	}
	if remote.calls != 1 || d.writes != writes {
		t.Fatal("replay renewed or rewrote participation")
	}
	q.ID = "another-controller"
	if _, err = b.Execute(t.Context(), p, q); err == nil {
		t.Fatal("another controller borrowed token")
	}
}

func TestPlan14ExecutionDeadlineFromRequestStart(t *testing.T) {
	for _, delay := range []time.Duration{30 * time.Second, 85 * time.Second, 100 * time.Second} {
		t.Run(delay.String(), func(t *testing.T) {
			b, _, remote, clock, q, p := brokerFixture(t)
			remote.onPut = func(int) { clock.advance(delay) }
			out, err := b.Execute(t.Context(), p, q)
			if delay >= 85*time.Second {
				if err == nil {
					t.Fatal("late response granted participation")
				}
				return
			}
			if err != nil {
				t.Fatal(err)
			}
			if out.SafeUntil != int64(185*time.Second) {
				t.Fatalf("response-relative deadline: %v", out)
			}
			clock.advance(55 * time.Second)
			q.Operation = "executionView"
			q.ID = "view-1"
			q.Token = out.Token
			if _, err = b.Execute(t.Context(), p, q); err == nil {
				t.Fatal("suspend beyond cutoff retained permission")
			}
		})
	}
}

func TestPlan14ExecutionPendingSurvivesCommitFailure(t *testing.T) {
	b, d, remote, _, q, p := brokerFixture(t)
	d.failAt = 2
	if _, err := b.Execute(t.Context(), p, q); err == nil {
		t.Fatal("failed commit replied successfully")
	}
	if remote.calls != 1 {
		t.Fatal("did not cross remote acquisition")
	}
	if _, err := b.Execute(t.Context(), p, q); err == nil {
		t.Fatal("uncertain commit replay granted permission")
	}
	if remote.calls != 1 {
		t.Fatal("uncertain replay acquired twice")
	}
	var record ExecutionRecord
	raw, err := d.Read(q.Host + ".json")
	if err != nil {
		t.Fatal(err)
	}
	if err = json.Unmarshal(raw, &record); err != nil {
		t.Fatal(err)
	}
	if record.Status == "settled" {
		t.Fatal("uncertain acquisition lost obligation")
	}
}

func TestPlan14ExecutionBrokerRestartCannotResume(t *testing.T) {
	b, _, _, _, q, p := brokerFixture(t)
	permit, err := b.Execute(t.Context(), p, q)
	if err != nil {
		t.Fatal(err)
	}
	if err := b.Close(t.Context()); err != nil {
		t.Fatal(err)
	}
	cfg := b.cfg
	cfg.Identity = "broker-2"
	restarted, err := NewRootBroker(cfg)
	if err != nil {
		t.Fatal(err)
	}
	defer restarted.Close(t.Context())
	q.Operation = "executionView"
	q.Token = permit.Token
	q.ID = "view-1"
	if _, err = restarted.Execute(t.Context(), p, q); !errors.Is(err, ErrLost) {
		t.Fatalf("restart restored prior permit: %v", err)
	}
}

func TestPlan14ExecutionAuthorityLossIsSticky(t *testing.T) {
	b, _, _, _, q, p := brokerFixture(t)
	out, err := b.Execute(t.Context(), p, q)
	if err != nil {
		t.Fatal(err)
	}
	original := b.cfg.Authority
	b.cfg.Authority = func(context.Context, ExecutionRequest, ControllerProcess) (ExecutionAuthority, error) {
		return ExecutionAuthority{}, ErrInactive
	}
	q.Operation = "executionView"
	q.Token = out.Token
	q.ID = "changed"
	if _, err = b.Execute(t.Context(), p, q); err == nil {
		t.Fatal("changed authority remained active")
	}
	b.cfg.Authority = original
	q.ID = "restored"
	if _, err = b.Execute(t.Context(), p, q); err == nil {
		t.Fatal("restoring authority resurrected lost lease")
	}
}

func TestPlan14ExecutionInvalidAuthorityAndReplayAreSticky(t *testing.T) {
	for _, name := range []string{"invalid", "replay"} {
		t.Run(name, func(t *testing.T) {
			replay := name == "replay"
			b, _, _, _, q, peer := brokerFixture(t)
			initial, err := b.Execute(t.Context(), peer, q)
			if err != nil {
				t.Fatal(err)
			}
			original := b.cfg.Authority
			b.cfg.Authority = func(ctx context.Context, q ExecutionRequest, p ControllerProcess) (ExecutionAuthority, error) {
				a, err := original(ctx, q, p)
				if replay {
					a.Phase = "changed"
				} else {
					a.Migration = 0
				}
				return a, err
			}
			if !replay {
				q.Operation, q.ID, q.Token = "executionView", "invalid-authority", initial.Token
			}
			if _, err := b.Execute(t.Context(), peer, q); err == nil {
				t.Fatal("invalid authority accepted")
			}
			b.cfg.Authority = original
			q.Operation, q.ID, q.Token = "executionView", "after-repair", initial.Token
			if _, err := b.Execute(t.Context(), peer, q); err == nil {
				t.Fatal("authority repair resurrected denied participation", replay)
			}
		})
	}
}

func TestPlan14ExecutionUncertainReleaseCanSettleAfterExpiry(t *testing.T) {
	b, d, remote, clock, q, p := brokerFixture(t)
	remote.onPut = func(int) { d.failAt = d.writes + 1 }
	if _, err := b.Execute(t.Context(), p, q); err == nil {
		t.Fatal("failed acquisition committed")
	}
	r, err := b.read(q.Host)
	if err != nil {
		t.Fatal(err)
	}
	// Simulate the crash window with no retained conditional-write response.
	r.LeaseETag = ""
	r.Status = "pending"
	if err = b.write(r); err != nil {
		t.Fatal(err)
	}
	q.Operation = "executionSettle"
	q.Token = r.Token
	q.ID = "settle"
	if _, err = b.Execute(t.Context(), p, q); err == nil {
		t.Fatal("unexpired uncertain lease settled")
	}
	clock.advance(TTL + time.Second)
	if _, err = b.Execute(t.Context(), p, q); err != nil {
		t.Fatal("expired exact obligation cannot settle", err)
	}
}

func TestPlan14ExecutionMonitorIndependentOfRenewal(t *testing.T) {
	b, _, remote, clock, q, p := brokerFixture(t)
	out, err := b.Execute(t.Context(), p, q)
	if err != nil {
		t.Fatal(err)
	}
	entered, releasePut := make(chan struct{}), make(chan struct{})
	remote.onPut = func(int) { close(entered); <-releasePut }
	cleaned := make(chan struct{}, 1)
	b.cfg.Cleanup = func(context.Context, ExecutionRecord) error {
		select {
		case cleaned <- struct{}{}:
		default:
		}
		return nil
	}
	done := make(chan error, 1)
	q.Operation = "executionRenew"
	q.ID = "renew"
	q.Token = out.Token
	go func() { _, err := b.Execute(context.Background(), p, q); done <- err }()
	<-entered
	clock.advance(TTL)
	monitored := make(chan error, 1)
	go func() { monitored <- b.Monitor(context.Background()) }()
	select {
	case <-cleaned:
	case <-time.After(time.Second):
		close(releasePut)
		<-done
		t.Fatal("network renewal blocked deadline cleanup")
	}
	close(releasePut)
	if err = <-done; err == nil {
		t.Fatal("late renewal revived expired lease")
	}
	<-monitored
}

func TestPlan14ExecutionBindsRegisteredInstallation(t *testing.T) {
	b, _, _, _, q, p := brokerFixture(t)
	base := b.cfg.Authority
	installed := false
	b.cfg.Authority = func(ctx context.Context, q ExecutionRequest, p ControllerProcess) (ExecutionAuthority, error) {
		a, e := base(ctx, q, p)
		if !installed {
			a.Installation = ""
			a.ParticipationOnly = true
		}
		return a, e
	}
	out, err := b.Execute(t.Context(), p, q)
	if err != nil {
		t.Fatal(err)
	}
	installed = true
	q.Operation = "executionView"
	q.ID = "registered"
	q.Token = out.Token
	if _, err = b.Execute(t.Context(), p, q); err != nil {
		t.Fatal("root registration lost participation", err)
	}
	r, err := b.read(q.Host)
	if err != nil {
		t.Fatal(err)
	}
	if r.Authority.Installation == "" || r.Authority.ParticipationOnly {
		t.Fatal("lease was not bound to root installation")
	}
	installed = false
	q.ID = "removed"
	if _, err = b.Execute(t.Context(), p, q); err == nil {
		t.Fatal("removed root installation retained participation")
	}
}

func TestPlan14ExecutionRootReadIgnoresMirrors(t *testing.T) {
	b, d, _, _, q, p := brokerFixture(t)
	original, oldPeer := b.cfg.Authority, p
	var err error
	p, err = currentControllerProcess()
	if err != nil {
		t.Fatal(err)
	}
	b.cfg.BootID = p.KernelBootID
	b.cfg.Authority = func(ctx context.Context, q ExecutionRequest, _ ControllerProcess) (ExecutionAuthority, error) {
		return original(ctx, q, oldPeer)
	}
	out, err := b.Execute(t.Context(), p, q)
	if err != nil {
		t.Fatal(err)
	}
	census, _ := json.Marshal(map[string]any{"Schema": 1, "Box": "box", "Enrollment": "en", "Migration": 1, "Revision": 1, "LaunchRevision": 1, "Complete": true})
	if err = d.Write("census.json", census); err != nil {
		t.Fatal(err)
	}
	if _, err = ReadRoot(t.Context(), d, b.cfg.Clock, q, false); err != nil {
		t.Fatal("valid root permit refused", err)
	}
	saved, err := b.read(q.Host)
	if err != nil {
		t.Fatal(err)
	}
	only := saved
	only.Authority.ParticipationOnly = true
	if err := b.write(only); err != nil {
		t.Fatal(err)
	}
	if _, err := ReadRoot(t.Context(), d, b.cfg.Clock, q, false); err == nil {
		t.Fatal("participation-only record granted privileged execution")
	}
	if err := b.write(saved); err != nil {
		t.Fatal(err)
	}
	data := t.TempDir()
	mirror := filepath.Join(data, ".hostexec")
	if err = os.Mkdir(mirror, 0700); err != nil {
		t.Fatal(err)
	}
	if err = os.WriteFile(filepath.Join(mirror, q.Host+".json"), []byte(`{"safe_until_boottime_ns":999999999999999999,"state":"active"}`), 0600); err != nil {
		t.Fatal(err)
	}
	if err = os.Rename(mirror, filepath.Join(data, "old")); err != nil {
		t.Fatal(err)
	}
	if err = os.Mkdir(mirror, 0700); err != nil {
		t.Fatal(err)
	}
	if _, err = ReadRoot(t.Context(), d, b.cfg.Clock, q, false); err != nil {
		t.Fatal("service replacement changed root permit", err)
	}
	// A failed durable loss still drops the root-owned live lock.
	d.failAt = d.writes + 1
	r, err := b.read(q.Host)
	if err != nil {
		t.Fatal(err)
	}
	if err = b.deny(t.Context(), r, ErrLost); err == nil {
		t.Fatal("lost lease reported success")
	}
	r.Status = "active"
	r.Token = out.Token
	raw, _ := json.Marshal(r)
	if err = os.WriteFile(filepath.Join(d.dir, q.Host+".json"), raw, 0600); err != nil {
		t.Fatal(err)
	}
	if _, err = ReadRoot(t.Context(), d, b.cfg.Clock, q, false); err == nil {
		t.Fatal("old active bytes revived revoked live lock")
	}
}

func TestPlan14ExecutionLossPersistenceRevokesAllLiveLocks(t *testing.T) {
	b, d, _, _, q, peer := brokerFixture(t)
	original := b.cfg.Authority
	b.cfg.Authority = func(ctx context.Context, request ExecutionRequest, peer ControllerProcess) (ExecutionAuthority, error) {
		request.Host = q.Host
		return original(ctx, request, peer)
	}
	if _, err := b.Execute(t.Context(), peer, q); err != nil {
		t.Fatal(err)
	}
	other := q
	other.Host = "other.example.com"
	if _, err := b.Execute(t.Context(), peer, other); err != nil {
		t.Fatal(err)
	}
	r, err := b.read(q.Host)
	if err != nil {
		t.Fatal(err)
	}
	d.failAt = d.writes + 1
	if err := b.deny(t.Context(), r, ErrLost); err == nil {
		t.Fatal("loss persistence failure accepted")
	}
	held, err := d.LiveHeld(other.Host + ".live")
	if err != nil || held {
		t.Fatal("independent reader still sees live permit", held, err)
	}

}

func TestPlan14ExecutionVisibleRenameCannotGrantBeforeSync(t *testing.T) {
	b, d, _, clock, q, priorPeer := brokerFixture(t)
	peer, err := currentControllerProcess()
	if err != nil {
		t.Fatal(err)
	}
	b.cfg.BootID = peer.KernelBootID
	original := b.cfg.Authority
	b.cfg.Authority = func(ctx context.Context, q ExecutionRequest, _ ControllerProcess) (ExecutionAuthority, error) {
		return original(ctx, q, priorPeer)
	}
	initial, err := b.Execute(t.Context(), peer, q)
	if err != nil {
		t.Fatal(err)
	}
	census, _ := json.Marshal(map[string]any{"Schema": 1, "Box": "box", "Enrollment": "en", "Migration": 1, "Revision": 1, "LaunchRevision": 1, "Complete": true})
	if err := d.Write("census.json", census); err != nil {
		t.Fatal(err)
	}
	if _, err := ReadRoot(t.Context(), d, clock, q, false); err != nil {
		t.Fatal("initial permit", err)
	}
	clock.advance(Renewal)
	visible, finish := make(chan struct{}), make(chan struct{})
	d.afterRename = func(_ string, raw []byte) {
		var r ExecutionRecord
		if json.Unmarshal(raw, &r) == nil && r.Status == "active" {
			close(visible)
			<-finish
		}
	}
	q.Operation, q.ID, q.Token = "executionRenew", "renew-before-sync", initial.Token
	done := make(chan error, 1)
	go func() { _, err := b.Execute(t.Context(), peer, q); done <- err }()
	select {
	case <-visible:
	case <-time.After(time.Second):
		t.Fatal("renew did not reach visible rename")
	}
	_, readErr := ReadRoot(t.Context(), d, clock, q, false)
	close(finish)
	if err := <-done; err != nil {
		t.Fatal(err)
	}
	if readErr == nil {
		t.Fatal("visible but unsynced active bytes granted execution")
	}
	if _, err := ReadRoot(t.Context(), d, clock, q, false); err != nil {
		t.Fatal("durable renewal refused", err)
	}
}

func TestPlan14ExecutionSettledTombstoneDoesNotBlockNextOperation(t *testing.T) {
	b, _, _, _, q, p := brokerFixture(t)
	out, err := b.Execute(t.Context(), p, q)
	if err != nil {
		t.Fatal(err)
	}
	settle := q
	settle.ID = "settle"
	settle.Operation = "executionSettle"
	settle.Token = out.Token
	if _, err = b.Execute(t.Context(), p, settle); err != nil {
		t.Fatal(err)
	}
	next := q
	next.ID = "next-begin"
	next.Op = "move-2"
	original := b.cfg.Authority
	b.cfg.Authority = func(ctx context.Context, got ExecutionRequest, p ControllerProcess) (ExecutionAuthority, error) {
		copy := got
		copy.Op = q.Op
		return original(ctx, copy, p)
	}
	granted, err := b.Execute(t.Context(), p, next)
	if err != nil {
		t.Fatal("settled operation permanently blocked host", err)
	}
	if granted.Token == out.Token {
		t.Fatal("new operation reused old permission")
	}
	if old, err := b.Execute(t.Context(), p, settle); err != nil || old.Status != "settled" {
		t.Fatalf("settlement tombstone lost: %+v %v", old, err)
	}
	if _, err = b.Execute(t.Context(), p, q); err == nil {
		t.Fatal("old begin resurrected")
	}
}

func TestPlan14ExecutionSessionTransitionRetainsRemoteOwner(t *testing.T) {
	b, _, remote, _, q, p := brokerFixture(t)
	original := b.cfg.Authority
	changed := false
	b.cfg.Authority = func(ctx context.Context, got ExecutionRequest, p ControllerProcess) (ExecutionAuthority, error) {
		expected := q.Session
		if changed {
			expected++
		}
		if got.Session != expected {
			return ExecutionAuthority{}, ErrInactive
		}
		copy := got
		copy.Session = q.Session
		a, e := original(ctx, copy, p)
		if changed {
			a.ClaimRevision = "claim-2"
			a.Phase = "configured"
		}
		return a, e
	}
	b.cfg.Transition = func(_ context.Context, old ExecutionRecord, next ExecutionRequest, a ExecutionAuthority) error {
		if old.Request.Session+1 != next.Session || old.Authority.Installation != a.Installation {
			return ErrLost
		}
		return nil
	}
	first, err := b.Execute(t.Context(), p, q)
	if err != nil {
		t.Fatal(err)
	}
	changed = true
	next := q
	next.ID = "new-session"
	next.Operation = "executionRenew"
	next.Session++
	next.Token = first.Token
	second, err := b.Execute(t.Context(), p, next)
	if err != nil {
		t.Fatal(err)
	}
	if second.Token == first.Token || second.Session != next.Session || remote.calls != 2 {
		t.Fatalf("transition: %+v calls=%d", second, remote.calls)
	}
	again, err := b.Execute(t.Context(), p, next)
	if err != nil || again != second || remote.calls != 2 {
		t.Fatalf("lost transition response replay: %+v %v", again, err)
	}
	stale := next
	stale.Operation = "executionView"
	stale.ID = "stale-view"
	if _, err = b.Execute(t.Context(), p, stale); err == nil {
		t.Fatal("old token admitted new session")
	}
}

type executionCrashStore struct{ release.Store }

func (s executionCrashStore) Put(ctx context.Context, key string, raw []byte, etag string) (string, error) {
	next, err := s.Store.Put(ctx, key, raw, etag)
	if err == nil {
		os.Exit(73)
	}
	return next, err
}

func TestPlan14ExecutionProcessCrashBeforeCommit(t *testing.T) {
	if dir := os.Getenv("PLAN14_EXECUTION_CRASH_DIR"); dir != "" {
		b, _, remote, _, q, p := brokerFixtureAt(t, filepath.Join(dir, "local"), filepath.Join(dir, "remote"))
		b.cfg.Remote = executionCrashStore{remote.Store}
		_, _ = b.Execute(context.Background(), p, q)
		t.Fatal("crash boundary was not reached")
	}
	dir := t.TempDir()
	for _, name := range []string{"local", "remote"} {
		if err := os.Mkdir(filepath.Join(dir, name), 0700); err != nil {
			t.Fatal(err)
		}
	}
	child := exec.Command(os.Args[0], "-test.run=^TestPlan14ExecutionProcessCrashBeforeCommit$", "-test.count=1")
	child.Dir = dir
	child.Env = append(os.Environ(), "PLAN14_EXECUTION_CRASH_DIR="+dir)
	raw, err := child.CombinedOutput()
	var exited *exec.ExitError
	if !errors.As(err, &exited) || exited.ExitCode() != 73 {
		t.Fatalf("child: %v %s", err, raw)
	}
	b, _, remote, clock, q, p := brokerFixtureAt(t, filepath.Join(dir, "local"), filepath.Join(dir, "remote"))
	b.cfg.Identity = "restarted-broker"
	record, err := b.read(q.Host)
	if err != nil {
		t.Fatal(err)
	}
	if record.Status != "pending" || record.LeaseETag != "" {
		t.Fatalf("crash did not retain uncertain pending: %+v", record)
	}
	key, _ := Key(q.Host, record.Authority.Box)
	if _, _, err = remote.Get(t.Context(), key); err != nil {
		t.Fatal("remote acquisition missing", err)
	}
	if _, err = b.Execute(t.Context(), p, q); err == nil {
		t.Fatal("crashed acquisition became a permit")
	}
	q.Operation = "executionSettle"
	q.ID = "settle"
	q.Token = record.Token
	if _, err = b.Execute(t.Context(), p, q); err == nil {
		t.Fatal("uncertain unexpired acquisition settled")
	}
	clock.advance(TTL + time.Second)
	if _, err = b.Execute(t.Context(), p, q); err != nil {
		t.Fatal("expired acquired obligation could not settle", err)
	}
}

func TestPlan14ExecutionRenewCannotCrossPriorCutoff(t *testing.T) {
	b, _, remote, clock, q, peer := brokerFixture(t)
	out, err := b.Execute(t.Context(), peer, q)
	if err != nil {
		t.Fatal(err)
	}
	clock.advance(80 * time.Second)
	remote.onPut = func(int) { clock.advance(6 * time.Second) }
	q.Operation, q.ID, q.Token = "executionRenew", "late-renew", out.Token
	if out, err := b.Execute(t.Context(), peer, q); err == nil {
		t.Fatalf("late renewal revived expired admission: %+v", out)
	}
}