fix: harden embedded Pi lifecycle recovery

This commit is contained in:
2026-08-04 23:14:47 +02:00
parent a368889838
commit 935bb1db0e
15 changed files with 572 additions and 101 deletions
+8 -4
View File
@@ -76,7 +76,7 @@ func Configure(ctx context.Context, runner Runner, value Defaults) error {
}
restore := func(cause error) error {
if restoreErr := restoreSettingsFile(context.Background(), runner, old); restoreErr != nil {
return fmt.Errorf("%w; previous Pi settings restoration could not be verified: recovery required", cause)
return fmt.Errorf("%w; previous Pi settings restoration could not be verified: %w", cause, restoreErr)
}
restoredEffective, restoreErr := readEffectiveSettings(context.Background(), runner)
if restoreErr != nil || !bytes.Equal(restoredEffective, oldEffective) {
@@ -155,12 +155,16 @@ func restoreSettingsFile(ctx context.Context, runner Runner, snapshot settingsFi
args := []string{"compose", "exec", "-T", "core", "node", "/app/backend/dist/settings/settings-cli.js", "--restore"}
result, restoreErr := runner.Run(ctx, args, bytes.NewReader(payload))
verified, verifyErr := captureSettingsFile(ctx, runner)
if restoreErr != nil {
cause := commandError("Pi installation settings restore", result, restoreErr)
if verifyErr == nil && verified == snapshot {
return recoveryRequired("previous Pi settings bytes were restored but durability was not acknowledged", cause)
}
return cause
}
if verifyErr == nil && verified == snapshot {
return nil
}
if restoreErr != nil {
return commandError("Pi installation settings restore", result, restoreErr)
}
return errors.New("Pi installation settings restore did not reproduce the exact prior file state")
}
+52 -7
View File
@@ -61,14 +61,56 @@ func TestConfigureRestoresAndVerifiesOldSettingsAfterEveryPostSnapshotFailure(t
}
}
func TestSettingsRestoreDoesNotMaskExplicitDurabilityFailureWithMatchingReadback(t *testing.T) {
old := Defaults{Provider: "old", Model: "old-model", Thinking: "low"}
raw, _ := json.Marshal(old)
fake := &configureRunner{
failure: "restore-durability",
settings: Defaults{Provider: "new", Model: "new-model", Thinking: "high"},
settingsExist: true,
settingsRaw: []byte(`{"provider":"new","model":"new-model","thinking":"high"}`),
}
snapshot := settingsFileSnapshot{Exists: true, RawBase64: base64.StdEncoding.EncodeToString(raw)}
err := restoreSettingsFile(context.Background(), fake, snapshot)
var recovery interface{ RecoveryRequired() bool }
if err == nil || !errors.As(err, &recovery) || !recovery.RecoveryRequired() {
t.Fatalf("restore error = %v; want typed recovery-required result", err)
}
if !fake.settingsExist || string(fake.settingsRaw) != string(raw) || fake.settings != old {
t.Fatalf("restored state = exists:%t raw:%q value:%#v; want exact old bytes", fake.settingsExist, fake.settingsRaw, fake.settings)
}
}
func TestConfigurePreservesTypedRecoveryRequiredErrorFromSettingsRestore(t *testing.T) {
old := Defaults{Provider: "old", Model: "old-model", Thinking: "low"}
raw, _ := json.Marshal(old)
fake := &configureRunner{
failure: "helper",
restoreDurabilityFailure: true,
settings: old,
settingsExist: true,
settingsRaw: raw,
}
err := Configure(context.Background(), fake, Defaults{Provider: "new", Model: "new-model", Thinking: "high"})
var recovery interface{ RecoveryRequired() bool }
if err == nil || !errors.As(err, &recovery) || !recovery.RecoveryRequired() {
t.Fatalf("Configure() error = %v; want typed recovery-required result", err)
}
}
type configureRunner struct {
failure string
settings Defaults
settingsExist bool
settingsRaw []byte
settingsReads int
configReads int
writes int
failure string
restoreDurabilityFailure bool
settings Defaults
settingsExist bool
settingsRaw []byte
settingsReads int
configReads int
writes int
}
func (f *configureRunner) Run(_ context.Context, args []string, stdin io.Reader) (compose.Result, error) {
@@ -103,6 +145,9 @@ func (f *configureRunner) Run(_ context.Context, args []string, stdin io.Reader)
if payload.Exists {
_ = json.Unmarshal(f.settingsRaw, &f.settings)
}
if f.failure == "restore-durability" || f.restoreDurabilityFailure {
return compose.Result{ExitCode: 2}, errors.New("injected post-rename directory fsync failure")
}
return compose.Result{}, nil
case strings.Contains(call, "settings-cli.js"):
if strings.Contains(call, "--provider new") {
@@ -0,0 +1,24 @@
package pi
// RecoveryRequiredError marks a result whose immediate state may be safe but whose durability
// was explicitly not acknowledged. Callers must not report success or clear maintenance.
type RecoveryRequiredError struct {
Operation string
Cause error
}
func (e *RecoveryRequiredError) Error() string {
return e.Operation + ": recovery required"
}
func (e *RecoveryRequiredError) Unwrap() error {
return e.Cause
}
func (e *RecoveryRequiredError) RecoveryRequired() bool {
return true
}
func recoveryRequired(operation string, cause error) error {
return &RecoveryRequiredError{Operation: operation, Cause: cause}
}
+133 -34
View File
@@ -289,10 +289,8 @@ func rollbackWithHooks(ctx context.Context, runner Runner, statePath string, con
if !confirm {
return Result{StatePath: statePath}, ErrConfirmationRequired
}
if err := setMaintenance(ctx, runner, true); err != nil {
return Result{StatePath: statePath}, err
}
clearMaintenance := true
maintenanceErr := ensureMaintenance(ctx, runner)
clearMaintenance := maintenanceErr == nil
defer func() {
if !clearMaintenance {
return
@@ -306,14 +304,18 @@ func rollbackWithHooks(ctx context.Context, runner Runner, statePath string, con
}
}
}()
if active, err := activeSessions(ctx, runner); err != nil {
return Result{StatePath: statePath}, err
} else if active {
return Result{StatePath: statePath}, ErrActiveSessions
if maintenanceErr == nil {
if active, err := activeSessions(ctx, runner); err != nil {
return Result{StatePath: statePath}, err
} else if active {
return Result{StatePath: statePath}, ErrActiveSessions
}
}
state, err := readState(statePath)
if err != nil {
clearMaintenance = false
if maintenanceErr == nil {
clearMaintenance = false
}
return Result{StatePath: statePath}, err
}
overridePath := lifecycleOverridePath(statePath, state.Transaction)
@@ -321,8 +323,17 @@ func rollbackWithHooks(ctx context.Context, runner Runner, statePath string, con
clearMaintenance = false
return Result{StatePath: statePath}, err
}
clearMaintenance = false
lifecycle := composeOverrideRunner{Runner: runner, path: overridePath}
if maintenanceErr != nil {
stopped, stopErr := coreIsStopped(ctx, runner)
if stopErr != nil || !stopped {
return Result{StatePath: statePath}, maintenanceErr
}
if err := persistMaintenanceWithoutLiveCore(ctx, lifecycle); err != nil {
return Result{Phase: PhaseFailed, StatePath: statePath}, err
}
}
clearMaintenance = false
if err := restore(ctx, lifecycle, state.Previous); err != nil {
state.Phase, state.Error = PhaseFailed, "rollback failed"
if writeErr := hooks.writeState(statePath, state); writeErr != nil {
@@ -330,8 +341,13 @@ func rollbackWithHooks(ctx context.Context, runner Runner, statePath string, con
}
return Result{Phase: PhaseFailed, StatePath: statePath}, err
}
if active, err := activeSessions(ctx, runner); err != nil {
return Result{Phase: PhaseFailed, StatePath: statePath}, err
} else if active {
return Result{Phase: PhaseFailed, StatePath: statePath}, ErrActiveSessions
}
if err := promoteLifecycleOverride(overridePath, currentImageOverridePath(statePath), state.Previous.Reference); err != nil {
return Result{Phase: PhaseFailed, StatePath: statePath}, errors.New("rollback restored the core but durable current-image promotion failed: recovery required")
return Result{Phase: PhaseFailed, StatePath: statePath}, fmt.Errorf("rollback restored the core but durable current-image promotion failed: %w", err)
}
state.Phase, state.Error = PhaseRolledBack, ""
if err := hooks.writeState(statePath, state); err != nil {
@@ -342,20 +358,27 @@ func rollbackWithHooks(ctx context.Context, runner Runner, statePath string, con
}
func compensate(ctx context.Context, runner Runner, statePath, overridePath string, state State, cause error, hooks lifecycleHooks) (Result, error, bool) {
if err := ensureMaintenance(context.Background(), runner); err != nil {
state.Phase, state.Error = PhaseFailed, "maintenance recovery failed"
_ = hooks.writeState(statePath, state)
return Result{Phase: PhaseFailed, StatePath: statePath}, errors.New("update failed and maintenance could not be reactivated: recovery required"), false
}
if active, err := activeSessions(context.Background(), runner); err != nil || active {
state.Phase, state.Error = PhaseFailed, "rollback inventory failed"
_ = hooks.writeState(statePath, state)
return Result{Phase: PhaseFailed, StatePath: statePath}, errors.New("update failed and rollback inventory is not quiescent: recovery required"), false
}
if err := writeLifecycleOverride(overridePath, state.Previous.Reference); err != nil {
return Result{Phase: PhaseFailed, StatePath: statePath}, errors.New("update failed and rollback override could not be prepared: recovery required"), false
}
lifecycle := composeOverrideRunner{Runner: runner, path: overridePath}
if err := ensureMaintenance(context.Background(), runner); err != nil {
stopped, stopErr := coreIsStopped(context.Background(), runner)
if stopErr != nil || !stopped {
state.Phase, state.Error = PhaseFailed, "maintenance recovery failed"
_ = hooks.writeState(statePath, state)
return Result{Phase: PhaseFailed, StatePath: statePath}, errors.New("update failed and maintenance could not be reactivated: recovery required"), false
}
if markerErr := persistMaintenanceWithoutLiveCore(context.Background(), lifecycle); markerErr != nil {
state.Phase, state.Error = PhaseFailed, "maintenance recovery failed"
_ = hooks.writeState(statePath, state)
return Result{Phase: PhaseFailed, StatePath: statePath}, errors.New("update failed and durable maintenance could not be established: recovery required"), false
}
} else if active, err := activeSessions(context.Background(), runner); err != nil || active {
state.Phase, state.Error = PhaseFailed, "rollback inventory failed"
_ = hooks.writeState(statePath, state)
return Result{Phase: PhaseFailed, StatePath: statePath}, errors.New("update failed and rollback inventory is not quiescent: recovery required"), false
}
if restoreErr := restore(ctx, lifecycle, state.Previous); restoreErr != nil {
state.Phase, state.Error = PhaseFailed, "candidate verification and automatic rollback failed"
if writeErr := hooks.writeState(statePath, state); writeErr != nil {
@@ -363,8 +386,13 @@ func compensate(ctx context.Context, runner Runner, statePath, overridePath stri
}
return Result{Phase: PhaseFailed, StatePath: statePath}, fmt.Errorf("update failed; automatic rollback also failed: recovery required"), false
}
if active, err := activeSessions(context.Background(), runner); err != nil || active {
state.Phase, state.Error = PhaseFailed, "restored rollback inventory failed"
_ = hooks.writeState(statePath, state)
return Result{Phase: PhaseFailed, StatePath: statePath}, errors.New("previous core image was restored but rollback inventory is not quiescent: recovery required"), false
}
if err := promoteLifecycleOverride(overridePath, currentImageOverridePath(statePath), state.Previous.Reference); err != nil {
return Result{Phase: PhaseFailed, StatePath: statePath}, errors.New("previous core image was restored but durable selector promotion failed: recovery required"), false
return Result{Phase: PhaseFailed, StatePath: statePath}, fmt.Errorf("previous core image was restored but durable selector promotion failed: %w", err), false
}
state.Phase, state.Error = PhaseRolledBack, ""
if writeErr := hooks.writeState(statePath, state); writeErr != nil {
@@ -373,6 +401,49 @@ func compensate(ctx context.Context, runner Runner, statePath, overridePath stri
return Result{Phase: PhaseRolledBack, StatePath: statePath}, fmt.Errorf("update failed; previous core image was restored: %w", cause), true
}
func coreIsStopped(ctx context.Context, runner Runner) (bool, error) {
result, err := runCompose(ctx, runner, "ps", "--status", "running", "-q", "core")
if err != nil {
return false, commandError("core running-state check", result, err)
}
return strings.TrimSpace(result.Stdout) == "", nil
}
const maintenanceMarkerScript = `
const fs = require("node:fs");
const path = require("node:path");
const marker = process.env.THT_MAINTENANCE_FILE;
if (!marker) throw new Error("THT_MAINTENANCE_FILE is required");
const directory = path.dirname(marker);
fs.mkdirSync(directory, { recursive: true });
const temporary = marker + ".rollback-" + process.pid + "-" + Date.now();
let file;
try {
file = fs.openSync(temporary, "wx", 0o600);
fs.writeFileSync(file, "{\"version\":1,\"active\":true}\n", "utf8");
fs.fsyncSync(file);
fs.closeSync(file);
file = undefined;
fs.renameSync(temporary, marker);
const directoryFile = fs.openSync(directory, "r");
try { fs.fsyncSync(directoryFile); } finally { fs.closeSync(directoryFile); }
} catch (error) {
if (file !== undefined) try { fs.closeSync(file); } catch {}
try { fs.unlinkSync(temporary); } catch {}
throw error;
}
`
func persistMaintenanceWithoutLiveCore(ctx context.Context, runner Runner) error {
result, err := runCompose(ctx, runner,
"run", "--rm", "--no-deps", "--entrypoint", "node", "core", "-e", maintenanceMarkerScript,
)
if err != nil {
return commandError("durable maintenance recovery", result, err)
}
return nil
}
func sourceValue(request Request) string {
if request.Source == PullSource {
return request.Image
@@ -403,15 +474,21 @@ func setMaintenance(ctx context.Context, runner Runner, enabled bool) error {
args := []string{"exec", "-T", "core", "curl", "-fsS", "-X", "POST", "http://127.0.0.1:8787/internal/maintenance/" + path}
result, err := runCompose(ctx, runner, args...)
status, valid := parseMaintenanceStatus(result.Stdout)
if err == nil && valid && status.Active == enabled && status.Admissions == 0 {
if err == nil && valid && status.Active == enabled && status.Admissions == 0 && !status.RecoveryRequired {
return nil
}
// A core recreate or transport interruption may lose only the response. Resolve ambiguity by
// reading the durable gate state before deciding that operator recovery is required.
observed, statusErr := MaintenanceStatus(ctx, runner)
if statusErr == nil && observed.Active == enabled && observed.Admissions == 0 {
if observed.RecoveryRequired {
return recoveryRequired("maintenance durability was explicitly not acknowledged", err)
}
return nil
}
if valid && status.RecoveryRequired {
return recoveryRequired("maintenance durability was explicitly not acknowledged", err)
}
if err != nil {
return commandError("maintenance admission gate", result, err)
}
@@ -419,8 +496,9 @@ func setMaintenance(ctx context.Context, runner Runner, enabled bool) error {
}
type MaintenanceState struct {
Active bool `json:"active"`
Admissions int `json:"admissions"`
Active bool `json:"active"`
Admissions int `json:"admissions"`
RecoveryRequired bool `json:"recoveryRequired"`
}
func parseMaintenanceStatus(value string) (MaintenanceState, bool) {
@@ -443,9 +521,12 @@ func MaintenanceStatus(ctx context.Context, runner Runner) (MaintenanceState, er
func ensureMaintenance(ctx context.Context, runner Runner) error {
status, err := MaintenanceStatus(ctx, runner)
if err == nil && status.Active && status.Admissions == 0 {
if err == nil && status.Active && status.Admissions == 0 && !status.RecoveryRequired {
return nil
}
if err == nil && status.RecoveryRequired {
return recoveryRequired("maintenance durability was explicitly not acknowledged", nil)
}
return setMaintenance(ctx, runner, true)
}
@@ -539,13 +620,9 @@ func verifyCandidate(ctx context.Context, runner Runner, wanted string, previous
if err != nil {
return commandError("core health check", health, err)
}
version, err := Status(ctx, runner)
if err != nil {
if err := verifyCandidateVersionIdentity(ctx, runner, wanted); err != nil {
return err
}
if version != wanted {
return errors.New("candidate Pi version does not match requested pinned version")
}
if err := Test(ctx, runner); err != nil {
return err
}
@@ -566,6 +643,21 @@ func verifyCandidate(ctx context.Context, runner Runner, wanted string, previous
return nil
}
func verifyCandidateVersionIdentity(ctx context.Context, runner Runner, wanted string) error {
executable, err := Status(ctx, runner)
if err != nil {
return err
}
environment, label, err := expectedVersions(ctx, runner)
if err != nil {
return err
}
if executable != wanted || environment != wanted || label != wanted {
return errors.New("candidate Pi executable, PI_VERSION, and image label do not all match the requested pinned version")
}
return nil
}
func restore(ctx context.Context, runner Runner, previous Image) error {
if err := tagImage(ctx, runner, previous.ID, previous.Reference, "rollback image restore"); err != nil {
return err
@@ -657,12 +749,19 @@ func writeLifecycleOverride(path, image string) error {
}
func promoteLifecycleOverride(source, destination, expectedImage string) error {
if err := durableReplace(source, destination, filepath.Dir(destination)); err != nil {
return promoteLifecycleOverrideWith(source, destination, expectedImage, durableReplace)
}
func promoteLifecycleOverrideWith(
source, destination, expectedImage string,
replace func(string, string, string) error,
) error {
if err := replace(source, destination, filepath.Dir(destination)); err != nil {
selected, readErr := readLifecycleOverride(destination)
if readErr == nil && selected == expectedImage {
return nil
return recoveryRequired("lifecycle image override changed but durability was not acknowledged", err)
}
return errors.New("lifecycle image override could not be promoted durably")
return recoveryRequired("lifecycle image override could not be promoted durably", err)
}
selected, err := readLifecycleOverride(destination)
if err != nil || selected != expectedImage {
+212 -2
View File
@@ -102,6 +102,38 @@ func TestSuccessfulUpdateAndRollbackRemainSelectedOnFreshRecreate(t *testing.T)
}
}
func TestSelectorPromotionDoesNotMaskPostRenameDirectoryFsyncFailure(t *testing.T) {
directory := t.TempDir()
source := filepath.Join(directory, "candidate.yaml")
destination := filepath.Join(directory, "current-image.yaml")
if err := writeLifecycleOverride(source, "thothii-core:candidate"); err != nil {
t.Fatal(err)
}
if err := writeLifecycleOverride(destination, "thothii-core:old"); err != nil {
t.Fatal(err)
}
err := promoteLifecycleOverrideWith(
source,
destination,
"thothii-core:candidate",
func(source, destination, _ string) error {
if err := os.Rename(source, destination); err != nil {
return err
}
return errors.New("injected post-rename directory fsync failure")
},
)
var recovery interface{ RecoveryRequired() bool }
if err == nil || !errors.As(err, &recovery) || !recovery.RecoveryRequired() {
t.Fatalf("promotion error = %v; want typed recovery-required result", err)
}
if selected := readSelectorReference(t, destination); selected != "thothii-core:candidate" {
t.Fatalf("immediate selector = %q, want landed candidate bytes", selected)
}
}
func TestTwoInstallationsSharingAConfiguredTagUseDifferentLifecycleTags(t *testing.T) {
first, second := newFakeRunner(), newFakeRunner()
firstPath := filepath.Join(t.TempDir(), "one", "state.json")
@@ -162,6 +194,21 @@ func TestMaintenanceLostResponsesAreResolvedByStatusAndEveryRecreateStartsGated(
}
}
func TestMaintenanceReconciliationDoesNotMaskExplicitDurabilityFailure(t *testing.T) {
fake := newFakeRunner()
fake.fail = "maintenance-activate-durability"
err := setMaintenance(context.Background(), fake, true)
var recovery interface{ RecoveryRequired() bool }
if err == nil || !errors.As(err, &recovery) || !recovery.RecoveryRequired() {
t.Fatalf("maintenance error = %v; want typed recovery-required result", err)
}
if !fake.maintenance {
t.Fatal("safe marker state was not retained after activation durability failure")
}
}
func TestCompensationReactivatesMaintenanceAndRescansBeforeRollback(t *testing.T) {
fake := newFakeRunner()
fake.fail = "version"
@@ -179,6 +226,63 @@ func TestCompensationReactivatesMaintenanceAndRescansBeforeRollback(t *testing.T
}
}
func TestAutomaticRollbackSurvivesADeadCandidateCore(t *testing.T) {
fake := newFakeRunner()
fake.fail = "dead-candidate"
statePath := filepath.Join(t.TempDir(), ".thothctl", "update-state.json")
result, err := Update(context.Background(), fake, Request{
StatePath: statePath,
Version: "0.81.0",
Source: BuildSource,
Confirm: true,
})
if err == nil || result.Phase != PhaseRolledBack {
t.Fatalf("Update() = %+v, %v; want automatic rollback after dead candidate", result, err)
}
if fake.currentImage != "sha256:old" || !fake.coreRunning {
t.Fatalf("restored core = image:%q running:%t; want previous running image", fake.currentImage, fake.coreRunning)
}
if fake.execFailuresWhileStopped == 0 {
t.Fatal("fake did not exercise candidate exec failure")
}
assertCalled(t, fake.calls, "run --rm --no-deps --entrypoint node")
assertMaintenanceClearedAfterRestoredProof(t, fake)
}
func TestManualRollbackSurvivesADeadCandidateCore(t *testing.T) {
fake := newFakeRunner()
statePath := filepath.Join(t.TempDir(), ".thothctl", "update-state.json")
previous := stateImageForTest(t, fake)
previous.Reference = "thothii-core:thothctl-dead-candidate-previous"
fake.tags[previous.Reference] = previous.ID
writeStateForTest(t, statePath, State{
Transaction: "dead-candidate",
Phase: PhaseRecreated,
MutationStarted: true,
Previous: previous,
})
fake.currentImage = "sha256:candidate"
fake.version = "0.81.0"
fake.coreRunning = false
fake.maintenance = false
result, err := Rollback(context.Background(), fake, statePath, true)
if err != nil || result.Phase != PhaseRolledBack {
t.Fatalf("Rollback() = %+v, %v; want restored previous core", result, err)
}
if fake.currentImage != "sha256:old" || !fake.coreRunning {
t.Fatalf("restored core = image:%q running:%t; want previous running image", fake.currentImage, fake.coreRunning)
}
if fake.execFailuresWhileStopped == 0 {
t.Fatal("fake did not exercise candidate exec failure")
}
assertCalled(t, fake.calls, "run --rm --no-deps --entrypoint node")
assertMaintenanceClearedAfterRestoredProof(t, fake)
}
func TestUpdatePullsOnlyDigestPinnedSource(t *testing.T) {
fake := newFakeRunner()
digest := "registry.example.invalid/thothii-core@sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
@@ -234,6 +338,36 @@ func TestUpdateRollsBackAfterPostRecreateFailures(t *testing.T) {
}
}
func TestCandidateVerificationRejectsEveryDeclaredVersionBoundaryMismatch(t *testing.T) {
for _, boundary := range []string{"executable", "environment", "image-label"} {
t.Run(boundary, func(t *testing.T) {
fake := newFakeRunner()
switch boundary {
case "executable":
fake.candidateVersion = "0.80.9"
case "environment":
fake.candidateExpectedVersion = "0.80.9"
case "image-label":
fake.candidateLabelVersion = "0.80.9"
}
result, err := Update(context.Background(), fake, Request{
StatePath: filepath.Join(t.TempDir(), "state.json"),
Version: "0.81.0",
Source: BuildSource,
Confirm: true,
})
if err == nil || result.Phase != PhaseRolledBack {
t.Fatalf("Update() = %+v, %v; want rollback for candidate %s mismatch", result, err, boundary)
}
if fake.currentImage != "sha256:old" {
t.Fatalf("current image = %q, want restored previous image", fake.currentImage)
}
})
}
}
func TestEveryRecoveryStateWriteFailureIsHandledTransactionally(t *testing.T) {
for failAt := 1; failAt <= 6; failAt++ {
t.Run(fmt.Sprintf("write-%d", failAt), func(t *testing.T) {
@@ -604,19 +738,35 @@ type fakeRunner struct {
dropMaintenanceAfterCandidate bool
modelsWire string
rollbackPrepared bool
coreRunning bool
execFailuresWhileStopped int
maintenanceHelperImages []string
maintenanceClearImages []string
restoredProofComplete bool
candidateVersion string
candidateExpectedVersion string
candidateLabelVersion string
}
func newFakeRunner() *fakeRunner {
return &fakeRunner{
version: "0.80.3", expectedVersion: "0.80.3", labelVersion: "0.80.3", currentImage: "sha256:old", configuredImage: "thothii-core:local",
tags: map[string]string{"thothii-core:local": "sha256:old"},
imageVersions: map[string]string{"sha256:old": "0.80.3"},
tags: map[string]string{"thothii-core:local": "sha256:old"},
imageVersions: map[string]string{"sha256:old": "0.80.3"},
coreRunning: true,
candidateVersion: "0.81.0",
candidateExpectedVersion: "0.81.0",
candidateLabelVersion: "0.81.0",
}
}
func (f *fakeRunner) Run(_ context.Context, args []string, _ io.Reader) (compose.Result, error) {
call := strings.Join(args, " ")
f.calls = append(f.calls, call)
if !f.coreRunning && containsArg(args, "exec") {
f.execFailuresWhileStopped++
return compose.Result{ExitCode: 1}, errors.New("core service is not running")
}
if f.built && f.fail != "compensation" && strings.Contains(call, "image tag sha256:old") {
f.fail = ""
}
@@ -651,6 +801,11 @@ func (f *fakeRunner) Run(_ context.Context, args []string, _ io.Reader) (compose
endpoint = "https://drift.example.invalid"
}
return compose.Result{Stdout: `{"services":{"core":{"image":"` + selectedCoreReference(args, f.configuredImage) + `","environment":{"THT_LLM_URL":"` + endpoint + `"}}}}`}, nil
case strings.Contains(call, "ps --status running -q core"):
if f.coreRunning {
return compose.Result{Stdout: "core-container\n"}, nil
}
return compose.Result{}, nil
case strings.Contains(call, "ps -q core"):
return compose.Result{Stdout: "core-container\n"}, nil
case strings.Contains(call, "inspect --format {{.Image}}"):
@@ -667,6 +822,9 @@ func (f *fakeRunner) Run(_ context.Context, args []string, _ io.Reader) (compose
return compose.Result{Stdout: `[{"Type":"volume","Name":"settings","Source":"settings","Destination":"/data/settings","RW":true},{"Type":"volume","Name":"pi-state","Source":"pi-state","Destination":"/home/thoth/.pi","RW":true},{"Type":"volume","Name":"sessions","Source":"sessions","Destination":"/data/sessions","RW":true},{"Type":"volume","Name":"workspace-registry","Source":"workspace-registry","Destination":"/data/workspace-registry","RW":true}]`}, nil
case strings.Contains(call, "/internal/maintenance/activate"):
f.maintenance = true
if f.fail == "maintenance-activate-durability" {
return compose.Result{ExitCode: 22}, errors.New("maintenance activation durability was not acknowledged")
}
if f.lostMaintenanceResponse == "activate" {
f.lostMaintenanceResponse = ""
return compose.Result{ExitCode: 52}, errors.New("lost activation response")
@@ -677,12 +835,16 @@ func (f *fakeRunner) Run(_ context.Context, args []string, _ io.Reader) (compose
return compose.Result{ExitCode: 53}, errors.New("maintenance clear failure")
}
f.maintenance = false
f.maintenanceClearImages = append(f.maintenanceClearImages, f.currentImage)
if f.lostMaintenanceResponse == "deactivate" {
f.lostMaintenanceResponse = ""
return compose.Result{ExitCode: 52}, errors.New("lost deactivation response")
}
return compose.Result{Stdout: `{"active":false,"admissions":0}`}, nil
case strings.Contains(call, "/internal/maintenance/status"):
if f.fail == "maintenance-activate-durability" {
return compose.Result{Stdout: fmt.Sprintf(`{"active":%t,"admissions":0,"recoveryRequired":true}`, f.maintenance)}, nil
}
return compose.Result{Stdout: fmt.Sprintf(`{"active":%t,"admissions":0}`, f.maintenance)}, nil
case strings.Contains(call, "/sessions?scope=all"):
if f.sessionsWire != "" {
@@ -693,6 +855,17 @@ func (f *fakeRunner) Run(_ context.Context, args []string, _ io.Reader) (compose
return compose.Result{Stdout: `[{"status":"open","archived":false}]`}, nil
}
return compose.Result{Stdout: `[]`}, nil
case containsArg(args, "run") && containsArg(args, "--entrypoint") && containsArg(args, "node"):
reference := selectedCoreReference(args, f.configuredImage)
f.maintenanceHelperImages = append(f.maintenanceHelperImages, reference)
if reference == "" {
return compose.Result{ExitCode: 1}, errors.New("maintenance helper has no selected image")
}
if f.tags[reference] != "sha256:old" {
return compose.Result{ExitCode: 1}, errors.New("maintenance helper did not select the previous image")
}
f.maintenance = true
return compose.Result{}, nil
case containsArg(args, "build"):
f.built = true
f.buildReference = selectedCoreReference(args, f.configuredImage)
@@ -723,6 +896,21 @@ func (f *fakeRunner) Run(_ context.Context, args []string, _ io.Reader) (compose
if version, ok := f.imageVersions[f.currentImage]; ok {
f.version = version
}
if f.currentImage == "sha256:candidate" {
f.version = f.candidateVersion
f.expectedVersion = f.candidateExpectedVersion
f.labelVersion = f.candidateLabelVersion
} else if f.currentImage == "sha256:old" {
f.expectedVersion = "0.80.3"
f.labelVersion = "0.80.3"
}
f.coreRunning = true
if f.currentImage == "sha256:candidate" {
f.restoredProofComplete = false
}
if f.fail == "dead-candidate" && f.currentImage == "sha256:candidate" {
f.coreRunning = false
}
if f.dropMaintenanceAfterCandidate && f.currentImage == "sha256:candidate" {
f.maintenance = false
f.dropMaintenanceAfterCandidate = false
@@ -744,6 +932,9 @@ func (f *fakeRunner) Run(_ context.Context, args []string, _ io.Reader) (compose
}
return compose.Result{Stdout: `{"models":[{"id":"model","provider":"provider"}]}`}, nil
case strings.Contains(call, "/settings"):
if f.currentImage == "sha256:old" {
f.restoredProofComplete = true
}
return compose.Result{Stdout: `{"provider":"provider","model":"model","thinking":"medium"}`}, nil
case strings.Contains(call, "/health"):
return compose.Result{Stdout: `{"status":"ok"}`}, nil
@@ -840,6 +1031,25 @@ func assertNotCalled(t *testing.T, calls []string, prohibited string) {
}
}
}
func assertMaintenanceClearedAfterRestoredProof(t *testing.T, fake *fakeRunner) {
t.Helper()
if fake.maintenance {
t.Fatal("maintenance remained active after restored-core proof")
}
if !fake.restoredProofComplete {
t.Fatal("maintenance cleared before restored settings smoke completed")
}
if len(fake.maintenanceClearImages) == 0 {
t.Fatal("maintenance was never durably cleared")
}
for _, image := range fake.maintenanceClearImages {
if image != "sha256:old" {
t.Fatalf("maintenance cleared while image %q was selected; want previous image", image)
}
}
}
func readStateBytes(t *testing.T, path string) []byte {
t.Helper()
contents, err := os.ReadFile(path)