diff --git a/docs/contracts/thothctl-pi.md b/docs/contracts/thothctl-pi.md new file mode 100644 index 00000000..8842842a --- /dev/null +++ b/docs/contracts/thothctl-pi.md @@ -0,0 +1,75 @@ +# `thothctl pi` lifecycle contract + +`thothctl` is the only component that drives Docker lifecycle operations. The `core` container +does not mount a Docker socket and Pi is never updated in a running container. + +## Read-only operations + +```text +thothctl --installation /absolute/path/thothii-installation.yaml pi status +thothctl --installation /absolute/path/thothii-installation.yaml pi doctor +thothctl --installation /absolute/path/thothii-installation.yaml pi test +thothctl --installation /absolute/path/thothii-installation.yaml pi logs [--follow] +``` + +`status` executes the image-bundled `pi --version`. `doctor` requires the rendered `core` image, +the external `THT_LLM_URL` contract, writable `/home/thoth/.pi`, the read-only Pi auth file, and +private `/health`. `test` additionally reads private `/models` and `/settings`; this temporary +composite smoke is replaced by the Pi Management API in Task 8. `logs` is core-only and uses the +same credential redaction as every other `thothctl` diagnostic. + +`pi check` is an alias for `pi test` for operational scripts. + +## Updating Pi + +An update always specifies a pinned Pi version and an explicit confirmation: + +```text +thothctl --installation /absolute/path/thothii-installation.yaml pi update \ + --version 0.81.0 --source build --yes +``` + +`--source build` rebuilds only `core` using `PI_VERSION=`. A pulled source must be a +digest-pinned image; tags are rejected: + +```text +thothctl --installation /absolute/path/thothii-installation.yaml pi update \ + --version 0.81.0 --source pull \ + --image registry.example.invalid/thothii-core@sha256:<64-lowercase-hex-digits> --yes +``` + +Before changing anything, the command validates the rendered Compose configuration, Pi auth/state +preconditions, health, current version, active sessions, the current image ID, named-volume set, +and a digest of the rendered non-secret configuration. It therefore keeps the exact installation +Compose files and environment, including the external `THT_LLM_URL` endpoint, when it recreates +only `core` with `--no-deps --force-recreate`. It never recreates `frontend` and never uses volume +replacement flags. + +Open, unarchived sessions stop an update. After an operator has completed or otherwise drained +their work, `--drain` makes the command re-check that the session list is empty before continuing. + +## Recovery and rollback + +Before a candidate is built or pulled, the command atomically writes: + +```text +/.thothctl/update-state.json +``` + +The file is mode `0600` and records only the previous/candidate image references and IDs, named +volume names, requested version/source, rendered-configuration digest, phase, and timestamp. It +never contains credentials, endpoint values, secret paths, Compose output, or logs. + +After recreate, the command checks core health, the requested `pi --version`, the Pi/core smoke, +unchanged configuration digest, and unchanged named-volume set. Any failure after recreation +automatically retags and recreates the recorded previous image. A failed or interrupted operation +leaves the same metadata for explicit operator recovery: + +```text +thothctl --installation /absolute/path/thothii-installation.yaml pi rollback --yes +``` + +Both update and rollback require `--yes`; without it they exit `2` before invoking Docker. Invalid +arguments, a pending recovery, and active sessions also exit `2`. Docker or verification failures +exit nonzero with concise, redacted guidance. The original Docker child exit code is preserved for +direct read-only/log command failures. diff --git a/tools/thothctl/cmd/thothctl/main.go b/tools/thothctl/cmd/thothctl/main.go index 9fda2422..cba2d972 100644 --- a/tools/thothctl/cmd/thothctl/main.go +++ b/tools/thothctl/cmd/thothctl/main.go @@ -15,6 +15,7 @@ import ( "github.com/aritmolab/thothii/tools/thothctl/internal/compose" "github.com/aritmolab/thothii/tools/thothctl/internal/config" "github.com/aritmolab/thothii/tools/thothctl/internal/output" + "github.com/aritmolab/thothii/tools/thothctl/internal/pi" ) const usage = `Usage: thothctl --installation /thothii-installation.yaml @@ -26,6 +27,12 @@ Commands: start Start the installation in the background. stop Stop the installation. update --check-only Validate the current installation without changing containers. + pi status Show the Pi version embedded in core. + pi doctor Check Pi preconditions without changing the installation. + pi test Run the temporary Pi/core smoke checks. + pi update Rebuild or pull a pinned Pi image (requires --yes). + pi rollback --yes Restore the image recorded by the latest Pi update. + pi logs [--follow] Show sanitized core logs. ` func main() { @@ -92,12 +99,152 @@ func run(ctx context.Context, args []string, stdout, stderr io.Writer) int { return commandUsageError(stderr, "doctor does not accept arguments") } return doctor(ctx, installation, runner, secretValues, stdout, stderr) + case "pi": + return piCommand(ctx, installation, runner, commandArgs, secretValues, stdout, stderr) default: return commandUsageError(stderr, fmt.Sprintf("unknown command %q", command)) } return writeResult(result, err, secretValues, stdout, stderr) } +// installationRunner transforms only Compose invocations into the installation's validated, +// profile-specific argument list. Direct Docker image commands remain host-side and use arguments. +type installationRunner struct { + installation config.Installation + runner compose.Runner +} + +func (r installationRunner) Run(ctx context.Context, args []string, stdin io.Reader) (compose.Result, error) { + if len(args) > 0 && args[0] == "compose" { + return r.runner.Run(ctx, r.installation.ComposeArgs(args[1:]...), stdin) + } + return r.runner.Run(ctx, args, stdin) +} + +func piCommand(ctx context.Context, installation config.Installation, runner compose.Runner, args []string, secretValues []string, stdout, stderr io.Writer) int { + if len(args) == 0 { + return commandUsageError(stderr, "pi requires a subcommand") + } + controlled := installationRunner{installation: installation, runner: runner} + switch args[0] { + case "status": + if len(args) != 1 { + return commandUsageError(stderr, "pi status does not accept arguments") + } + version, err := pi.Status(ctx, controlled) + if err != nil { + return piFailure(stderr, err, secretValues) + } + fmt.Fprintf(stdout, "Pi version: %s\n", output.Sanitize(version, secretValues)) + return 0 + case "doctor": + if len(args) != 1 { + return commandUsageError(stderr, "pi doctor does not accept arguments") + } + if err := pi.Doctor(ctx, controlled); err != nil { + return piFailure(stderr, err, secretValues) + } + fmt.Fprintln(stdout, "Pi preflight checks passed.") + return 0 + case "test", "check": + if len(args) != 1 { + return commandUsageError(stderr, "pi test does not accept arguments") + } + if err := pi.Test(ctx, controlled); err != nil { + return piFailure(stderr, err, secretValues) + } + fmt.Fprintln(stdout, "Pi/core smoke checks passed.") + return 0 + case "logs": + logArgs, err := logsArgs(args[1:]) + if err != nil { + return commandUsageError(stderr, "pi logs accepts only --follow") + } + logArgs = append(logArgs, "core") + result, err := controlled.Run(ctx, append([]string{"compose"}, logArgs...), nil) + return writeResult(result, err, secretValues, stdout, stderr) + case "update": + request, err := parsePiUpdateArgs(args[1:], filepath.Join(installation.ProjectDirectory, ".thothctl", "update-state.json")) + if err != nil { + return commandUsageError(stderr, err.Error()) + } + result, err := pi.Update(ctx, controlled, request) + if err != nil { + return piFailure(stderr, err, secretValues) + } + if result.Phase == pi.PhaseNoop { + fmt.Fprintf(stdout, "Pi already runs requested version %s; no container was recreated.\n", request.Version) + return 0 + } + fmt.Fprintf(stdout, "Pi update verified. Recovery metadata: %s\n", result.StatePath) + return 0 + case "rollback": + if len(args) != 2 || args[1] != "--yes" { + return commandUsageError(stderr, "pi rollback requires --yes") + } + result, err := pi.Rollback(ctx, controlled, filepath.Join(installation.ProjectDirectory, ".thothctl", "update-state.json"), true) + if err != nil { + return piFailure(stderr, err, secretValues) + } + fmt.Fprintf(stdout, "Pi rollback restored the recorded core image. Recovery metadata: %s\n", result.StatePath) + return 0 + default: + return commandUsageError(stderr, fmt.Sprintf("unknown pi command %q", args[0])) + } +} + +func parsePiUpdateArgs(args []string, statePath string) (pi.Request, error) { + request := pi.Request{StatePath: statePath, Source: pi.BuildSource} + for len(args) > 0 { + switch args[0] { + case "--version": + if len(args) < 2 || request.Version != "" { + return pi.Request{}, errors.New("pi update requires one --version ") + } + request.Version, args = args[1], args[2:] + case "--source": + if len(args) < 2 { + return pi.Request{}, errors.New("--source requires build or pull") + } + request.Source, args = pi.Source(args[1]), args[2:] + case "--image": + if len(args) < 2 || request.Image != "" { + return pi.Request{}, errors.New("--image requires one digest-pinned image reference") + } + request.Image, args = args[1], args[2:] + case "--yes": + if request.Confirm { + return pi.Request{}, errors.New("--yes may be supplied once") + } + request.Confirm, args = true, args[1:] + case "--drain": + if request.Drain { + return pi.Request{}, errors.New("--drain may be supplied once") + } + request.Drain, args = true, args[1:] + default: + return pi.Request{}, fmt.Errorf("unknown pi update option %q", args[0]) + } + } + if request.Version == "" { + return pi.Request{}, errors.New("pi update requires --version ") + } + return request, nil +} + +func piFailure(stderr io.Writer, err error, secretValues []string) int { + code := 1 + if errors.Is(err, pi.ErrConfirmationRequired) || errors.Is(err, pi.ErrActiveSessions) || errors.Is(err, pi.ErrInterruptedUpdate) { + code = 2 + } + var childExit interface{ ExitCode() int } + if errors.As(err, &childExit) && childExit.ExitCode() != 0 { + code = childExit.ExitCode() + } + fmt.Fprintf(stderr, "thothctl: %s\n", output.Sanitize(err.Error(), secretValues)) + return code +} + func parseArgs(args []string) (string, string, []string, error) { if len(args) < 3 || args[0] != "--installation" { return "", "", nil, errors.New("--installation is required before the command") diff --git a/tools/thothctl/cmd/thothctl/main_test.go b/tools/thothctl/cmd/thothctl/main_test.go index c3c3a2fb..64e26d16 100644 --- a/tools/thothctl/cmd/thothctl/main_test.go +++ b/tools/thothctl/cmd/thothctl/main_test.go @@ -347,6 +347,59 @@ func TestRunPreservesChildExitCodes(t *testing.T) { } } +func TestRunPiStatusUsesImageBundledPi(t *testing.T) { + fixture := newCLIFixture(t, "THT_LLM_URL=https://llm.example.invalid\n") + fixture.setEnvironment(t) + + var stdout, stderr bytes.Buffer + exitCode := run(context.Background(), []string{"--installation", fixture.installationPath, "pi", "status"}, &stdout, &stderr) + + if exitCode != 0 { + t.Fatalf("run() exit code = %d, stderr = %s", exitCode, stderr.String()) + } + if stdout.String() != "Pi version: 0.80.3\n" { + t.Errorf("stdout = %q, want image-bundled Pi version", stdout.String()) + } + assertInvocationContains(t, fixture.invocations(t), "exec", "-T", "core", "pi", "--version") +} + +func TestRunPiUpdateRequiresExplicitConfirmationWithoutInvokingDocker(t *testing.T) { + fixture := newCLIFixture(t, "THT_LLM_URL=https://llm.example.invalid\n") + fixture.setEnvironment(t) + + var stdout, stderr bytes.Buffer + exitCode := run(context.Background(), []string{"--installation", fixture.installationPath, "pi", "update", "--version", "0.81.0"}, &stdout, &stderr) + + if exitCode != 2 { + t.Errorf("run() exit code = %d, want 2", exitCode) + } + if !strings.Contains(stderr.String(), "requires --yes") { + t.Errorf("stderr = %q, want explicit confirmation guidance", stderr.String()) + } + assertDockerNotInvoked(t, fixture) +} + +func TestRunPiStatusPreservesDockerExitCodeAndRedactsDiagnostics(t *testing.T) { + fixture := newCLIFixture(t, "PI_TOKEN_FILE=%s\n") + secretPath := filepath.Join(fixture.root, "pi-secret") + if err := os.WriteFile(secretPath, []byte("pi-status-secret"), 0o600); err != nil { + t.Fatal(err) + } + fixture.setEnvironment(t, secretPath) + t.Setenv("THOTHCTL_FAKE_FAIL_ON", "version") + t.Setenv("THOTHCTL_FAKE_FAILURE", "pi-status-secret") + + var stdout, stderr bytes.Buffer + exitCode := run(context.Background(), []string{"--installation", fixture.installationPath, "pi", "status"}, &stdout, &stderr) + + if exitCode != 41 { + t.Errorf("run() exit code = %d, want 41", exitCode) + } + if strings.Contains(stdout.String()+stderr.String(), "pi-status-secret") { + t.Errorf("Pi status exposed a secret: stdout=%q stderr=%q", stdout.String(), stderr.String()) + } +} + type cliFixture struct { root string installationPath string @@ -392,8 +445,9 @@ func newCLIFixture(t *testing.T, envTemplate string) cliFixture { printf '%s\n' "$@" >> "$THOTHCTL_FAKE_ARGS" printf '%s\n' -- >> "$THOTHCTL_FAKE_ARGS" case " $* " in - *" config --format json "*) printf '%s\n' '{"volumes":{"settings":{}}}' ;; + *" config --format json "*) printf '%s\n' '{"volumes":{"settings":{}},"services":{"core":{"image":"thothii-core:local","environment":{"THT_LLM_URL":"https://llm.example.invalid"}}}}' ;; *" ps --format json "*) printf '%s\n' '[{"Service":"core","State":"running","Health":"healthy"},{"Service":"frontend","State":"running","Health":"healthy"}]' ;; + *" pi --version "*) printf '%s\n' '0.80.3' ;; *" logs "*) printf '%s\n' "$THOTHCTL_FAKE_LOG" ;; esac if [ "${THOTHCTL_FAKE_FAIL_ON:-}" = "version" ]; then @@ -458,3 +512,18 @@ func assertDockerNotInvoked(t *testing.T, fixture cliFixture) { t.Errorf("Docker was invoked: stat error = %v", err) } } + +func assertInvocationContains(t *testing.T, invocations [][]string, want ...string) { + t.Helper() + for _, invocation := range invocations { + for start := range invocation { + if len(invocation)-start < len(want) { + continue + } + if strings.Join(invocation[start:start+len(want)], "\x00") == strings.Join(want, "\x00") { + return + } + } + } + t.Fatalf("invocations = %#v, want %#v", invocations, want) +} diff --git a/tools/thothctl/internal/pi/commands.go b/tools/thothctl/internal/pi/commands.go new file mode 100644 index 00000000..b98a5667 --- /dev/null +++ b/tools/thothctl/internal/pi/commands.go @@ -0,0 +1,120 @@ +package pi + +import ( + "context" + "crypto/sha256" + "encoding/json" + "errors" + "fmt" + "io" + "strings" + + "github.com/aritmolab/thothii/tools/thothctl/internal/compose" +) + +// Runner is the narrow, shell-free command boundary shared with thothctl. +type Runner interface { + Run(context.Context, []string, io.Reader) (compose.Result, error) +} + +// Status reports the image-bundled Pi version without using a host Pi executable. +func Status(ctx context.Context, runner Runner) (string, error) { + result, err := runCompose(ctx, runner, "exec", "-T", "core", "pi", "--version") + if err != nil { + return "", commandError("Pi version check", result, err) + } + version := strings.TrimSpace(result.Stdout) + if version == "" { + return "", errors.New("Pi version check returned no version") + } + return version, nil +} + +// Doctor verifies the installation-side invariants Pi needs before an update. +func Doctor(ctx context.Context, runner Runner) error { + if _, err := renderedCore(ctx, runner); err != nil { + return err + } + if _, err := Status(ctx, runner); err != nil { + return err + } + for _, check := range [][]string{ + {"exec", "-T", "core", "sh", "-ceu", "test -w /home/thoth/.pi"}, + {"exec", "-T", "core", "sh", "-ceu", "test -r /home/thoth/.pi/agent/auth.json"}, + {"exec", "-T", "core", "curl", "-fsS", "http://127.0.0.1:8787/health"}, + } { + result, err := runCompose(ctx, runner, check...) + if err != nil { + return commandError("Pi preflight check", result, err) + } + } + return nil +} + +// Test performs the pre-Task-8 composite smoke through core's private loopback endpoint. +func Test(ctx context.Context, runner Runner) error { + if _, err := Status(ctx, runner); err != nil { + return err + } + for _, path := range []string{"health", "models", "settings"} { + result, err := runCompose(ctx, runner, "exec", "-T", "core", "curl", "-fsS", "http://127.0.0.1:8787/"+path) + if err != nil { + return commandError("Pi smoke check", result, err) + } + var payload any + if err := json.Unmarshal([]byte(result.Stdout), &payload); err != nil { + return fmt.Errorf("Pi smoke check returned invalid %s response", path) + } + if _, ok := payload.(map[string]any); !ok { + return fmt.Errorf("Pi smoke check returned invalid %s response", path) + } + } + return nil +} + +func renderedCore(ctx context.Context, runner Runner) (Image, error) { + result, err := runCompose(ctx, runner, "config", "--format", "json") + if err != nil { + return Image{}, commandError("Compose configuration check", result, err) + } + var document struct { + Services map[string]struct { + Image string `json:"image"` + Environment map[string]any `json:"environment"` + } `json:"services"` + } + if err := json.Unmarshal([]byte(result.Stdout), &document); err != nil { + return Image{}, errors.New("Compose returned invalid rendered configuration") + } + core, exists := document.Services["core"] + if !exists || core.Image == "" { + return Image{}, errors.New("rendered Compose configuration has no core image") + } + endpoint, exists := core.Environment["THT_LLM_URL"].(string) + if !exists || strings.TrimSpace(endpoint) == "" { + return Image{}, errors.New("THT_LLM_URL must be configured before Pi lifecycle operations") + } + digest := sha256.Sum256([]byte(result.Stdout)) + return Image{Reference: core.Image, ConfigurationSHA: fmt.Sprintf("%x", digest[:])}, nil +} + +func runCompose(ctx context.Context, runner Runner, args ...string) (compose.Result, error) { + return runner.Run(ctx, append([]string{"compose"}, args...), nil) +} + +func commandError(label string, result compose.Result, err error) error { + if result.ExitCode != 0 { + return commandFailure{message: fmt.Sprintf("%s failed (exit %d)", label, result.ExitCode), exitCode: result.ExitCode} + } + return commandFailure{message: fmt.Sprintf("%s failed", label)} +} + +type commandFailure struct { + message string + exitCode int +} + +func (e commandFailure) Error() string { return e.message } + +// ExitCode exposes a Docker child exit code without exposing its output. +func (e commandFailure) ExitCode() int { return e.exitCode } diff --git a/tools/thothctl/internal/pi/commands_test.go b/tools/thothctl/internal/pi/commands_test.go new file mode 100644 index 00000000..98890185 --- /dev/null +++ b/tools/thothctl/internal/pi/commands_test.go @@ -0,0 +1,30 @@ +package pi + +import ( + "context" + "strings" + "testing" +) + +func TestDoctorRequiresExternalEndpointAuthPiStateAndHealth(t *testing.T) { + fake := newFakeRunner() + if err := Doctor(context.Background(), fake); err != nil { + t.Fatalf("Doctor() error = %v", err) + } + for _, command := range []string{"pi --version", "test -w /home/thoth/.pi", "test -r /home/thoth/.pi/agent/auth.json", "/health"} { + assertCalled(t, fake.calls, command) + } +} + +func TestTestUsesOnlySanitizedPiAndCoreProbes(t *testing.T) { + fake := newFakeRunner() + if err := Test(context.Background(), fake); err != nil { + t.Fatalf("Test() error = %v", err) + } + for _, command := range []string{"pi --version", "/health", "/models", "/settings"} { + assertCalled(t, fake.calls, command) + } + if got := strings.Join(fake.calls, "\n"); strings.Contains(got, "secret") { + t.Fatalf("probe commands expose secret: %s", got) + } +} diff --git a/tools/thothctl/internal/pi/state.go b/tools/thothctl/internal/pi/state.go new file mode 100644 index 00000000..d09abc22 --- /dev/null +++ b/tools/thothctl/internal/pi/state.go @@ -0,0 +1,106 @@ +// Package pi implements host-side lifecycle operations for the Pi bundled in core. +package pi + +import ( + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "time" +) + +const stateFileVersion = 1 + +// Phase describes the durable point reached by a Pi update. +type Phase string + +const ( + PhasePreflight Phase = "preflight" + PhaseBuilding Phase = "building" + PhaseRecreated Phase = "recreated" + PhaseVerified Phase = "verified" + PhaseRolledBack Phase = "rolled_back" + PhaseFailed Phase = "failed" + PhaseNoop Phase = "noop" +) + +// Image is the non-secret recovery identity of a core image and its mounted volume names. +type Image struct { + ID string `json:"id"` + Reference string `json:"reference"` + Volumes []string `json:"volumes"` + ConfigurationSHA string `json:"configuration_sha256,omitempty"` +} + +// Target records the immutable input selected by the operator. Source is either build or a +// digest-pinned image reference; it intentionally never contains credentials. +type Target struct { + Version string `json:"version"` + Source string `json:"source"` +} + +// State is recovery metadata stored below the installation project. It never stores environment +// values, secret paths, credentials, or command output. +type State struct { + Version int `json:"version"` + Phase Phase `json:"phase"` + UpdatedAt time.Time `json:"updated_at"` + Target Target `json:"target,omitempty"` + Previous Image `json:"previous"` + Candidate Image `json:"candidate,omitempty"` + Error string `json:"error,omitempty"` +} + +func readState(path string) (State, error) { + contents, err := os.ReadFile(path) + if err != nil { + return State{}, err + } + var state State + if err := json.Unmarshal(contents, &state); err != nil { + return State{}, errors.New("update recovery state is invalid") + } + if state.Version != stateFileVersion || state.Previous.ID == "" || state.Previous.Reference == "" { + return State{}, errors.New("update recovery state is incomplete") + } + return state, nil +} + +func writeState(path string, state State) error { + if state.Previous.ID == "" || state.Previous.Reference == "" { + return errors.New("refusing to write incomplete update recovery state") + } + state.Version = stateFileVersion + state.UpdatedAt = time.Now().UTC() + contents, err := json.MarshalIndent(state, "", " ") + if err != nil { + return fmt.Errorf("encode update recovery state: %w", err) + } + contents = append(contents, '\n') + directory := filepath.Dir(path) + if err := os.MkdirAll(directory, 0o700); err != nil { + return errors.New("could not create update recovery directory") + } + temporary, err := os.CreateTemp(directory, ".update-state-*.tmp") + if err != nil { + return errors.New("could not write update recovery state") + } + temporaryName := temporary.Name() + defer os.Remove(temporaryName) + if err := temporary.Chmod(0o600); err != nil { + temporary.Close() + return errors.New("could not protect update recovery state") + } + if _, err := temporary.Write(contents); err != nil { + temporary.Close() + return errors.New("could not write update recovery state") + } + if err := temporary.Close(); err != nil { + return errors.New("could not write update recovery state") + } + if err := os.Rename(temporaryName, path); err != nil { + return errors.New("could not finalize update recovery state") + } + return nil +} diff --git a/tools/thothctl/internal/pi/update.go b/tools/thothctl/internal/pi/update.go new file mode 100644 index 00000000..3300fca8 --- /dev/null +++ b/tools/thothctl/internal/pi/update.go @@ -0,0 +1,318 @@ +package pi + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "regexp" + "sort" + "strings" +) + +var ( + ErrConfirmationRequired = errors.New("update requires --yes after reviewing the planned Pi version") + ErrActiveSessions = errors.New("active sessions must be drained before updating Pi; use --drain only after they are complete") + ErrInterruptedUpdate = errors.New("a previous Pi update is incomplete; run pi rollback --yes before starting another update") + versionPattern = regexp.MustCompile(`^[0-9]+(?:\.[0-9]+){1,3}(?:[-+][0-9A-Za-z.-]+)?$`) + digestPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._/:@-]*@sha256:[a-f0-9]{64}$`) +) + +// Source chooses whether the candidate is built from this checkout or pulled from an immutable image. +type Source string + +const ( + BuildSource Source = "build" + PullSource Source = "pull" +) + +// Request contains only non-secret operator inputs. +type Request struct { + StatePath string + Version string + Source Source + Image string + Confirm bool + Drain bool +} + +// Result summarizes the completed, failed, or recovered transaction without command output. +type Result struct { + Phase Phase + StatePath string +} + +// Update performs a recoverable core-only Pi update using the default Compose command layout. +func Update(ctx context.Context, runner Runner, request Request) (Result, error) { + if request.StatePath == "" { + return Result{}, errors.New("update state path is required") + } + if !request.Confirm { + return Result{StatePath: request.StatePath}, ErrConfirmationRequired + } + if !versionPattern.MatchString(request.Version) { + return Result{StatePath: request.StatePath}, errors.New("Pi version must be an explicit pinned version") + } + if request.Source == "" { + request.Source = BuildSource + } + if request.Source != BuildSource && request.Source != PullSource { + return Result{StatePath: request.StatePath}, errors.New("Pi update source must be build or pull") + } + if request.Source == PullSource && !digestPattern.MatchString(request.Image) { + return Result{StatePath: request.StatePath}, errors.New("pulled Pi image must use an immutable sha256 digest") + } + if old, err := readState(request.StatePath); err == nil && old.Phase != PhaseVerified && old.Phase != PhaseRolledBack && old.Phase != PhaseNoop { + return Result{StatePath: request.StatePath}, ErrInterruptedUpdate + } else if err != nil && !errors.Is(err, os.ErrNotExist) { + return Result{StatePath: request.StatePath}, err + } + + running, err := activeSessions(ctx, runner) + if err != nil { + return Result{StatePath: request.StatePath}, err + } + if running { + if !request.Drain { + return Result{StatePath: request.StatePath}, ErrActiveSessions + } + running, err = activeSessions(ctx, runner) + if err != nil { + return Result{StatePath: request.StatePath}, err + } + if running { + return Result{StatePath: request.StatePath}, ErrActiveSessions + } + } + if err := Doctor(ctx, runner); err != nil { + return Result{StatePath: request.StatePath}, err + } + + currentVersion, err := Status(ctx, runner) + if err != nil { + return Result{StatePath: request.StatePath}, err + } + if currentVersion == request.Version { + return Result{Phase: PhaseNoop, StatePath: request.StatePath}, nil + } + configured, err := renderedCore(ctx, runner) + if err != nil { + return Result{StatePath: request.StatePath}, err + } + previous, err := runningImage(ctx, runner, configured.Reference) + if err != nil { + return Result{StatePath: request.StatePath}, err + } + previous.ConfigurationSHA = configured.ConfigurationSHA + state := State{Phase: PhasePreflight, Target: Target{Version: request.Version, Source: sourceValue(request)}, Previous: previous} + if err := writeState(request.StatePath, state); err != nil { + return Result{StatePath: request.StatePath}, err + } + + state.Phase = PhaseBuilding + if err := writeState(request.StatePath, state); err != nil { + return Result{StatePath: request.StatePath}, err + } + if err := prepareCandidate(ctx, runner, request, previous.Reference); err != nil { + state.Phase, state.Error = PhaseFailed, "candidate image preparation failed" + _ = writeState(request.StatePath, state) + return Result{Phase: PhaseFailed, StatePath: request.StatePath}, err + } + if err := recreateCore(ctx, runner); err != nil { + state.Phase, state.Error = PhaseFailed, "core recreation failed" + _ = writeState(request.StatePath, state) + return Result{Phase: PhaseFailed, StatePath: request.StatePath}, err + } + state.Phase = PhaseRecreated + state.Candidate, _ = runningImage(ctx, runner, previous.Reference) + if err := writeState(request.StatePath, state); err != nil { + return Result{Phase: PhaseRecreated, StatePath: request.StatePath}, err + } + if err := verifyCandidate(ctx, runner, request.Version, previous.Volumes, previous.ConfigurationSHA); err != nil { + return rollbackAfterFailure(ctx, runner, request.StatePath, state, err) + } + state.Phase, state.Error = PhaseVerified, "" + if err := writeState(request.StatePath, state); err != nil { + return Result{Phase: PhaseVerified, StatePath: request.StatePath}, err + } + return Result{Phase: PhaseVerified, StatePath: request.StatePath}, nil +} + +// Rollback restores the image recorded in durable update state. It is safe for interrupted runs. +func Rollback(ctx context.Context, runner Runner, statePath string, confirm bool) (Result, error) { + if !confirm { + return Result{StatePath: statePath}, ErrConfirmationRequired + } + state, err := readState(statePath) + if err != nil { + return Result{StatePath: statePath}, err + } + if err := restore(ctx, runner, state.Previous); err != nil { + state.Phase, state.Error = PhaseFailed, "rollback failed" + _ = writeState(statePath, state) + return Result{Phase: PhaseFailed, StatePath: statePath}, err + } + state.Phase, state.Error = PhaseRolledBack, "" + if err := writeState(statePath, state); err != nil { + return Result{Phase: PhaseRolledBack, StatePath: statePath}, err + } + return Result{Phase: PhaseRolledBack, StatePath: statePath}, nil +} + +func rollbackAfterFailure(ctx context.Context, runner Runner, statePath string, state State, cause error) (Result, error) { + if restoreErr := restore(ctx, runner, state.Previous); restoreErr != nil { + state.Phase, state.Error = PhaseFailed, "candidate verification and automatic rollback failed" + _ = writeState(statePath, state) + return Result{Phase: PhaseFailed, StatePath: statePath}, fmt.Errorf("candidate verification failed; automatic rollback also failed") + } + state.Phase, state.Error = PhaseRolledBack, "" + _ = writeState(statePath, state) + return Result{Phase: PhaseRolledBack, StatePath: statePath}, fmt.Errorf("candidate verification failed; previous core image was restored") +} + +func sourceValue(request Request) string { + if request.Source == PullSource { + return request.Image + } + return string(BuildSource) +} + +func activeSessions(ctx context.Context, runner Runner) (bool, error) { + result, err := runCompose(ctx, runner, "exec", "-T", "core", "tht", "session", "list", "--json") + if err != nil { + return false, commandError("active-session check", result, err) + } + var sessions []struct { + Status string `json:"status"` + Archived bool `json:"archived"` + } + if err := json.Unmarshal([]byte(result.Stdout), &sessions); err != nil { + return false, errors.New("active-session check returned invalid session data") + } + for _, session := range sessions { + if !session.Archived && session.Status == "open" { + return true, nil + } + } + return false, nil +} + +func runningImage(ctx context.Context, runner Runner, reference string) (Image, error) { + container, err := runCompose(ctx, runner, "ps", "-q", "core") + if err != nil || strings.TrimSpace(container.Stdout) == "" { + return Image{}, commandError("running core image check", container, err) + } + id := strings.TrimSpace(container.Stdout) + image, err := runner.Run(ctx, []string{"inspect", "--format", "{{.Image}}", id}, nil) + if err != nil || strings.TrimSpace(image.Stdout) == "" { + return Image{}, commandError("running core image check", image, err) + } + mounts, err := runner.Run(ctx, []string{"inspect", "--format", "{{range .Mounts}}{{if eq .Type \"volume\"}}{{.Name}}{{\"\\n\"}}{{end}}{{end}}", id}, nil) + if err != nil { + return Image{}, commandError("core volume check", mounts, err) + } + volumes := nonEmptyLines(mounts.Stdout) + if len(volumes) == 0 { + return Image{}, errors.New("core has no named volumes to preserve") + } + return Image{ID: strings.TrimSpace(image.Stdout), Reference: reference, Volumes: volumes}, nil +} + +func prepareCandidate(ctx context.Context, runner Runner, request Request, reference string) error { + if request.Source == BuildSource { + result, err := runCompose(ctx, runner, "build", "--pull", "--build-arg", "PI_VERSION="+request.Version, "core") + if err != nil { + return commandError("Pi image build", result, err) + } + return nil + } + pull, err := runner.Run(ctx, []string{"pull", request.Image}, nil) + if err != nil { + return commandError("Pi image pull", pull, err) + } + tag, err := runner.Run(ctx, []string{"image", "tag", request.Image, reference}, nil) + if err != nil { + return commandError("Pi image tag", tag, err) + } + return nil +} + +func recreateCore(ctx context.Context, runner Runner) error { + result, err := runCompose(ctx, runner, "up", "--detach", "--no-deps", "--force-recreate", "core") + if err != nil { + return commandError("core recreation", result, err) + } + return nil +} + +func verifyCandidate(ctx context.Context, runner Runner, wanted string, previousVolumes []string, configurationSHA string) error { + health, err := runCompose(ctx, runner, "exec", "-T", "core", "curl", "-fsS", "http://127.0.0.1:8787/health") + if err != nil { + return commandError("core health check", health, err) + } + version, err := Status(ctx, runner) + if 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 + } + configured, err := renderedCore(ctx, runner) + if err != nil { + return err + } + if configured.ConfigurationSHA != configurationSHA { + return errors.New("external endpoint configuration changed during Pi update") + } + after, err := runningImage(ctx, runner, configured.Reference) + if err != nil { + return err + } + if !sameStrings(previousVolumes, after.Volumes) { + return errors.New("core volume set changed during Pi update") + } + return nil +} + +func restore(ctx context.Context, runner Runner, previous Image) error { + tag, err := runner.Run(ctx, []string{"image", "tag", previous.ID, previous.Reference}, nil) + if err != nil { + return commandError("rollback image restore", tag, err) + } + if err := recreateCore(ctx, runner); err != nil { + return err + } + configured, err := renderedCore(ctx, runner) + if err != nil { + return err + } + after, err := runningImage(ctx, runner, configured.Reference) + if err != nil { + return err + } + if !sameStrings(previous.Volumes, after.Volumes) { + return errors.New("core volume set changed during rollback") + } + return nil +} + +func nonEmptyLines(text string) []string { + var values []string + for _, value := range strings.Split(text, "\n") { + if value = strings.TrimSpace(value); value != "" { + values = append(values, value) + } + } + sort.Strings(values) + return values +} +func sameStrings(left, right []string) bool { + left, right = append([]string(nil), left...), append([]string(nil), right...) + sort.Strings(left) + sort.Strings(right) + return strings.Join(left, "\x00") == strings.Join(right, "\x00") +} diff --git a/tools/thothctl/internal/pi/update_test.go b/tools/thothctl/internal/pi/update_test.go new file mode 100644 index 00000000..5ed917bd --- /dev/null +++ b/tools/thothctl/internal/pi/update_test.go @@ -0,0 +1,256 @@ +package pi + +import ( + "context" + "errors" + "io" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/aritmolab/thothii/tools/thothctl/internal/compose" +) + +func TestUpdateBuildsPinnedVersionRecreatesOnlyCoreAndPersistsRecoveryState(t *testing.T) { + fake := newFakeRunner() + dir := t.TempDir() + result, err := Update(context.Background(), fake, Request{ + StatePath: filepath.Join(dir, ".thothctl", "update-state.json"), + Version: "0.81.0", + Source: BuildSource, + Confirm: true, + }) + if err != nil { + t.Fatalf("Update() error = %v", err) + } + if result.Phase != PhaseVerified { + t.Fatalf("phase = %q, want %q", result.Phase, PhaseVerified) + } + assertCalled(t, fake.calls, "compose build --pull --build-arg PI_VERSION=0.81.0 core") + assertCalled(t, fake.calls, "compose up --detach --no-deps --force-recreate core") + assertNotCalled(t, fake.calls, "frontend") + if got := string(readStateBytes(t, result.StatePath)); strings.Contains(got, "secret") || !strings.Contains(got, `"phase": "verified"`) { + t.Fatalf("state = %q, want credential-free verified metadata", got) + } + if got := string(readStateBytes(t, result.StatePath)); strings.Contains(got, "llm.example.invalid") { + t.Fatalf("state = %q, want an endpoint-free configuration digest", got) + } +} + +func TestUpdatePullsOnlyDigestPinnedSource(t *testing.T) { + fake := newFakeRunner() + digest := "registry.example.invalid/thothii-core@sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + _, err := Update(context.Background(), fake, Request{StatePath: filepath.Join(t.TempDir(), "state.json"), Version: "0.81.0", Source: PullSource, Image: digest, Confirm: true}) + if err == nil { + t.Fatal("Update() error = nil, want Pi version verification failure from unchanged fake image") + } + assertCalled(t, fake.calls, "pull "+digest) + assertCalled(t, fake.calls, "image tag "+digest+" thothii-core:local") + + fake = newFakeRunner() + _, err = Update(context.Background(), fake, Request{StatePath: filepath.Join(t.TempDir(), "state.json"), Version: "0.81.0", Source: PullSource, Image: "registry.example.invalid/thothii-core:latest", Confirm: true}) + if err == nil || !strings.Contains(err.Error(), "immutable sha256 digest") { + t.Fatalf("Update() error = %v, want digest-pinning rejection", err) + } + assertNotCalled(t, fake.calls, "compose") +} + +func TestUpdateIsNoOpWhenDesiredVersionAlreadyRuns(t *testing.T) { + fake := newFakeRunner() + fake.version = "0.80.3" + result, err := Update(context.Background(), fake, Request{StatePath: filepath.Join(t.TempDir(), "state.json"), Version: "0.80.3", Source: BuildSource, Confirm: true}) + if err != nil { + t.Fatalf("Update() error = %v", err) + } + if result.Phase != PhaseNoop { + t.Fatalf("phase = %q, want %q", result.Phase, PhaseNoop) + } + assertNotCalled(t, fake.calls, "compose build") +} + +func TestUpdateRollsBackAfterPostRecreateFailures(t *testing.T) { + for _, failure := range []string{"health", "version", "smoke"} { + t.Run(failure, func(t *testing.T) { + fake := newFakeRunner() + fake.fail = failure + statePath := filepath.Join(t.TempDir(), "state.json") + result, err := Update(context.Background(), fake, Request{StatePath: statePath, Version: "0.81.0", Source: BuildSource, Confirm: true}) + if err == nil { + t.Fatal("Update() error = nil, want verification failure") + } + if result.Phase != PhaseRolledBack { + t.Fatalf("phase = %q, want %q", result.Phase, PhaseRolledBack) + } + assertCalled(t, fake.calls, "image tag sha256:old thothii-core:local") + assertCalled(t, fake.calls, "compose up --detach --no-deps --force-recreate core") + if strings.Join(fake.volumes, ",") != "settings,pi-state,sessions,workspace-registry" { + t.Fatalf("volumes changed: %v", fake.volumes) + } + if got := string(readStateBytes(t, statePath)); !strings.Contains(got, `"phase": "rolled_back"`) { + t.Fatalf("state = %q, want rollback metadata", got) + } + }) + } +} + +func TestUpdateDoesNotRecreateWhenPreflightOrBuildFails(t *testing.T) { + for _, failure := range []string{"preflight", "build"} { + t.Run(failure, func(t *testing.T) { + fake := newFakeRunner() + fake.fail = failure + 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 { + t.Fatal("Update() error = nil, want failure") + } + if result.Phase == PhaseRolledBack { + t.Fatalf("pre-recreate failure unexpectedly rolled back: %+v", result) + } + assertNotCalled(t, fake.calls, "force-recreate") + }) + } +} + +func TestUpdateRequiresConfirmationAndDrainsActiveSessions(t *testing.T) { + fake := newFakeRunner() + _, err := Update(context.Background(), fake, Request{StatePath: filepath.Join(t.TempDir(), "state.json"), Version: "0.81.0", Source: BuildSource}) + if !errors.Is(err, ErrConfirmationRequired) { + t.Fatalf("Update() error = %v, want confirmation error", err) + } + assertNotCalled(t, fake.calls, "compose") + + fake = newFakeRunner() + fake.activeSessions = true + _, err = Update(context.Background(), fake, Request{StatePath: filepath.Join(t.TempDir(), "state.json"), Version: "0.81.0", Source: BuildSource, Confirm: true}) + if !errors.Is(err, ErrActiveSessions) { + t.Fatalf("Update() error = %v, want active-sessions error", err) + } + + fake = newFakeRunner() + fake.activeSessions = true + _, err = Update(context.Background(), fake, Request{StatePath: filepath.Join(t.TempDir(), "state.json"), Version: "0.81.0", Source: BuildSource, Confirm: true, Drain: true}) + if err != nil { + t.Fatalf("Update() with drain error = %v", err) + } + assertCalled(t, fake.calls, "compose exec -T core tht session list --json") +} + +func TestRollbackRestoresInterruptedOrPreviouslyRecordedState(t *testing.T) { + fake := newFakeRunner() + statePath := filepath.Join(t.TempDir(), "state.json") + writeStateForTest(t, statePath, State{Version: 1, Phase: PhaseRecreated, Previous: Image{ID: "sha256:old", Reference: "thothii-core:local", Volumes: []string{"settings", "pi-state", "sessions", "workspace-registry"}}}) + result, err := Rollback(context.Background(), fake, statePath, true) + if err != nil { + t.Fatalf("Rollback() error = %v", err) + } + if result.Phase != PhaseRolledBack { + t.Fatalf("phase = %q, want %q", result.Phase, PhaseRolledBack) + } + assertCalled(t, fake.calls, "image tag sha256:old thothii-core:local") + assertCalled(t, fake.calls, "compose up --detach --no-deps --force-recreate core") +} + +func TestUpdateRefusesToOverwriteInterruptedRecoveryState(t *testing.T) { + fake := newFakeRunner() + statePath := filepath.Join(t.TempDir(), "state.json") + writeStateForTest(t, statePath, State{Phase: PhaseRecreated, Previous: Image{ID: "sha256:old", Reference: "thothii-core:local", Volumes: []string{"settings"}}}) + _, err := Update(context.Background(), fake, Request{StatePath: statePath, Version: "0.81.0", Source: BuildSource, Confirm: true}) + if !errors.Is(err, ErrInterruptedUpdate) { + t.Fatalf("Update() error = %v, want interrupted update error", err) + } + assertNotCalled(t, fake.calls, "compose") +} + +type fakeRunner struct { + calls []string + fail string + version string + activeSessions bool + built bool + volumes []string +} + +func newFakeRunner() *fakeRunner { + return &fakeRunner{version: "0.80.3", volumes: []string{"settings", "pi-state", "sessions", "workspace-registry"}} +} + +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.fail == "preflight" && strings.Contains(call, "config --format json") { + return compose.Result{ExitCode: 1}, errors.New("provider token=secret") + } + if f.fail == "build" && strings.Contains(call, "compose build") { + return compose.Result{ExitCode: 1}, errors.New("build token=secret") + } + if f.fail == "health" && f.built && strings.Contains(call, "curl -fsS http://127.0.0.1:8787/health") { + return compose.Result{ExitCode: 1}, errors.New("health token=secret") + } + if f.fail == "version" && f.built && strings.Contains(call, "pi --version") && strings.Contains(call, "exec") { + return compose.Result{ExitCode: 1}, errors.New("version token=secret") + } + if f.fail == "smoke" && strings.Contains(call, "curl -fsS http://127.0.0.1:8787/models") { + return compose.Result{ExitCode: 1}, errors.New("smoke token=secret") + } + switch { + case strings.Contains(call, "config --format json"): + return compose.Result{Stdout: `{"services":{"core":{"image":"thothii-core:local","environment":{"THT_LLM_URL":"https://llm.example.invalid"}}}}`}, nil + case strings.Contains(call, "ps -q core"): + return compose.Result{Stdout: "core-container\n"}, nil + case strings.Contains(call, "inspect --format {{.Image}}"): + return compose.Result{Stdout: "sha256:old\n"}, nil + case strings.Contains(call, "inspect --format {{range .Mounts}}"): + return compose.Result{Stdout: strings.Join(f.volumes, "\n") + "\n"}, nil + case strings.Contains(call, "tht session list --json"): + if f.activeSessions { + f.activeSessions = false + return compose.Result{Stdout: `[{"status":"open","archived":false}]`}, nil + } + return compose.Result{Stdout: "[]"}, nil + case strings.Contains(call, "compose build"): + f.built = true + f.version = "0.81.0" + return compose.Result{}, nil + case strings.Contains(call, "pi --version"): + return compose.Result{Stdout: f.version + "\n"}, nil + case strings.Contains(call, "/models"): + return compose.Result{Stdout: `{"models":[{"id":"model","provider":"provider"}]}`}, nil + case strings.Contains(call, "/settings"): + return compose.Result{Stdout: `{"provider":"provider","model":"model","thinking":"medium"}`}, nil + case strings.Contains(call, "/health"): + return compose.Result{Stdout: `{"status":"ok"}`}, nil + } + return compose.Result{}, nil +} + +func assertCalled(t *testing.T, calls []string, want string) { + t.Helper() + for _, call := range calls { + if strings.Contains(call, want) { + return + } + } + t.Fatalf("calls %v did not include %q", calls, want) +} +func assertNotCalled(t *testing.T, calls []string, prohibited string) { + t.Helper() + for _, call := range calls { + if strings.Contains(call, prohibited) { + t.Fatalf("calls %v unexpectedly included %q", calls, prohibited) + } + } +} +func readStateBytes(t *testing.T, path string) []byte { + t.Helper() + contents, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + return contents +} +func writeStateForTest(t *testing.T, path string, state State) { + t.Helper() + if err := writeState(path, state); err != nil { + t.Fatal(err) + } +}