feat: manage embedded pi with thothctl

This commit is contained in:
2026-08-04 18:20:39 +02:00
parent c90299f24d
commit 20e59b8d32
8 changed files with 1122 additions and 1 deletions
+75
View File
@@ -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=<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
<projectDirectory>/.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.
+147
View File
@@ -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 <absolute-path>/thothii-installation.yaml <command>
@@ -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 <pinned-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 <pinned-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 <absolute-path> is required before the command")
+70 -1
View File
@@ -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)
}
+120
View File
@@ -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 }
@@ -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)
}
}
+106
View File
@@ -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
}
+318
View File
@@ -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")
}
+256
View File
@@ -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)
}
}