From 2b26ea5e9615d9aa135fed77a33f0a05c87149c7 Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Wed, 30 Sep 2026 21:05:03 +0000 Subject: [PATCH] Move sandbox reset and address bindings into deployment and deploymentpg The reset becomes deployment use cases on ExecutionOperations: start or escalate, cancel, advance the auto deadline and complete, each a Load, Decide and Apply transaction on the execution lease. DeploymentTx gains the reset writes, the stored starter source and RecordAuditAs, so the deadline and completion audit as the administrator who started the reset. Completion in use is deployment.InUseError. The Worker lists the Sessions a reset still has to archive through the required Dispatcher.DeploymentReader; the per-Session archive stays in store. Address bindings are deployment.Service.AddressBindings, read in one snapshot against the public URL the service was built with, and serve the installation page. api maps reset and in-use errors in writeSandboxError, and the OpenAPI schemas become deployment.ResetRequest and deployment.AddressBindings. --- contracts/agents-api/core.openapi.yaml | 62 ++--- services/core/IMPLEMENTATION.md | 2 +- services/core/cmd/server/main.go | 4 +- services/core/internal/api/errors.go | 12 +- .../core/internal/api/errors_deployment.go | 4 + services/core/internal/api/fakes_test.go | 8 +- services/core/internal/api/installation.go | 8 +- .../core/internal/api/installation_test.go | 5 +- .../api/sandbox_deployment_changes_test.go | 61 +++-- .../internal/api/sandbox_deployment_setup.go | 22 +- services/core/internal/deployment/errors.go | 9 + .../core/internal/deployment/fakes_test.go | 83 ++++++- .../core/internal/deployment/public_url.go | 17 ++ services/core/internal/deployment/reset.go | 186 +++++++++++++++ .../core/internal/deployment/reset_test.go | 161 +++++++++++++ services/core/internal/deployment/storage.go | 30 ++- .../archive_cancellation_cleanup_test.go | 2 +- .../execution/deployment_fixture_test.go | 45 ++-- .../deployment_provider_observations_test.go | 2 +- .../core/internal/execution/dispatcher.go | 3 + .../core/internal/execution/owner_test.go | 40 ++-- .../internal/execution/runtime_lifecycle.go | 4 +- .../internal/execution/runtime_manager.go | 1 + .../runtime_retirement_failure_test.go | 2 +- .../sandbox_deployment_drain_test.go | 14 +- .../sandbox_deployment_setup_test.go | 12 +- .../sandbox_deployment_switch_test.go | 8 +- .../execution/sandbox_generations_test.go | 4 +- .../sandbox_provider_contract_test.go | 2 +- .../core/internal/execution/sandbox_reset.go | 12 +- .../internal/execution/sandbox_reset_test.go | 50 ++-- .../execution/sandbox_snapshot_budget_test.go | 9 +- services/core/internal/execution/worker.go | 5 +- .../postgres/deploymentpg/reset_test.go | 172 ++++++++++++++ .../postgres/deploymentpg/store.go | 41 ++++ .../persistence/postgres/deploymentpg/tx.go | 53 +++++ .../providers/configuration_flow_test.go | 4 +- .../internal/store/admin_session_archive.go | 3 +- .../core/internal/store/fixture_db_test.go | 7 +- .../store/public_handler_fixture_test.go | 4 +- services/core/internal/store/public_url.go | 28 --- .../core/internal/store/runtime_deployment.go | 5 + .../store/runtime_node_capacity_test.go | 2 +- .../sandbox_deployment_resources_test.go | 4 +- .../store/sandbox_deployment_switch_test.go | 18 +- .../sandbox_deployment_switch_worker_test.go | 2 +- services/core/internal/store/sandbox_reset.go | 223 ------------------ .../core/internal/store/sandbox_reset_test.go | 70 +++--- .../store/sandbox_specification_store_test.go | 7 +- 49 files changed, 1029 insertions(+), 503 deletions(-) create mode 100644 services/core/internal/deployment/reset.go create mode 100644 services/core/internal/deployment/reset_test.go create mode 100644 services/core/internal/persistence/postgres/deploymentpg/reset_test.go delete mode 100644 services/core/internal/store/sandbox_reset.go diff --git a/contracts/agents-api/core.openapi.yaml b/contracts/agents-api/core.openapi.yaml index 85bbd0005..46e9e53ab 100644 --- a/contracts/agents-api/core.openapi.yaml +++ b/contracts/agents-api/core.openapi.yaml @@ -429,7 +429,7 @@ definitions: api.Installation: properties: address_bindings: - $ref: '#/definitions/store.AddressBindings' + $ref: '#/definitions/deployment.AddressBindings' api_base_url: description: public_url followed by /v1; null when public_url is null. type: string @@ -902,6 +902,17 @@ definitions: service: $ref: '#/definitions/coremetrics.ServiceState' type: object + deployment.AddressBindings: + properties: + hosted_sandboxes: + type: integer + nodes: + type: integer + nodes_on_other_address: + type: integer + self_hosted_executors: + type: integer + type: object deployment.HostHistory: properties: points: @@ -1150,6 +1161,24 @@ definitions: on_offline_nodes: type: integer type: object + deployment.ResetRequest: + properties: + clear: + enum: + - auto + - force + type: string + deadline_seconds: + maximum: 86400 + minimum: 300 + type: integer + expected_generation: + minimum: 0 + type: integer + required: + - clear + - expected_generation + type: object deployment.Resources: properties: allocations: @@ -1364,17 +1393,6 @@ definitions: state: type: string type: object - store.AddressBindings: - properties: - hosted_sandboxes: - type: integer - nodes: - type: integer - nodes_on_other_address: - type: integer - self_hosted_executors: - type: integer - type: object store.AdminAssetCounts: properties: agents: @@ -1419,24 +1437,6 @@ definitions: tenant_id: type: string type: object - store.SandboxResetRequest: - properties: - clear: - enum: - - auto - - force - type: string - deadline_seconds: - maximum: 86400 - minimum: 300 - type: integer - expected_generation: - minimum: 0 - type: integer - required: - - clear - - expected_generation - type: object v1.Agent: properties: id: @@ -6326,7 +6326,7 @@ paths: name: body required: true schema: - $ref: '#/definitions/store.SandboxResetRequest' + $ref: '#/definitions/deployment.ResetRequest' produces: - application/json responses: diff --git a/services/core/IMPLEMENTATION.md b/services/core/IMPLEMENTATION.md index 8772f79c1..22c31b5b7 100644 --- a/services/core/IMPLEMENTATION.md +++ b/services/core/IMPLEMENTATION.md @@ -33,7 +33,7 @@ Domain owners, each with its PostgreSQL adapter under `internal/persistence/post - `environmenttemplates` (`templatepg`): Environment Templates, their validation and default network, their sealed setup, initial files, Skills and Plugins, and the resolved Template that Session creation composes into its Environment. - `modelconfiguration` (`modelconfigurationpg`): each Harness's deployment default model configuration and its last-use observations. - `skills` (`skillpg`): Skills and their immutable versions: archive checks, the default and latest pointers, version selection and deletion, and each version's sealed archive. Session creation freezes selected versions inside its `store` transaction with the `skills` rules. -- `deployment` (`deploymentpg`): the sandbox deployment and its nodes: provider configuration and the sealed credential, the specification and retained generations, setup, update and switch, node enrollment, identity and authentication, generation configuration, capacity, presence, status and host history. It interprets Sandbox Provider declarations through the `providers.Registry` it is given, whose lookups return typed errors. `cmd/server` builds that registry and calls it only to build a direct Provider and to discover a Provider's configuration; it takes each setup's mode and declared operations from the `Setup` that `deployment` returns. The only other reader is `store`, whose own `providers.Builtin()` validates the stored specification at Session admission and reads the public-origin requirement at placement. Deployment changes run through `deployment.ExecutionOperations` on `deploymentpg.NewExecution(lease, …)`, which the Worker receives as `execution.Owner.Deployment`; the Worker reads the deployment and prepares a selection's setup through the pooled `deployment.Service` in `execution.Dispatcher.Deployment`, and node management and reads use the pooled `deploymentpg.Store`, which `cmd/server` also reads the owner epoch from and runs the host-history sampler on. Provider calls run outside transactions, and the final transaction rechecks the expected generation. Allocations, placement and reset stay in `store`. +- `deployment` (`deploymentpg`): the sandbox deployment and its nodes: provider configuration and the sealed credential, the specification and retained generations, setup, update and switch, node enrollment, identity and authentication, generation configuration, capacity, presence, status, host history, reset, and the counts of nodes and sandboxes bound to the public address. It interprets Sandbox Provider declarations through the `providers.Registry` it is given, whose lookups return typed errors. `cmd/server` builds that registry and calls it only to build a direct Provider and to discover a Provider's configuration; it takes each setup's mode and declared operations from the `Setup` that `deployment` returns. The only other reader is `store`, whose own `providers.Builtin()` validates the stored specification at Session admission and reads the public-origin requirement at placement. Deployment changes run through `deployment.ExecutionOperations` on `deploymentpg.NewExecution(lease, …)`, which the Worker receives as `execution.Owner.Deployment`; the Worker reads the deployment and prepares a selection's setup through the pooled `deployment.Service` in `execution.Dispatcher.Deployment` and lists the Sessions a reset still has to archive through the pooled `deployment.Reader` in `execution.Dispatcher.DeploymentReader`, and node management and reads use the pooled `deploymentpg.Store`, which `cmd/server` also reads the owner epoch from and runs the host-history sampler on. Provider calls run outside transactions, and the final transaction rechecks the expected generation. Allocations, placement, reset admission and the reset's Session archive stay in `store`. - `sessions` (`sessionpg`): Session use cases and reads as they leave `store`. The pooled `sessions.Service` runs the use cases on `sessionpg.Store`, which implements `sessions.Storage`, and plain reads use `sessions.Reader`, which `sessionpg.Store` also implements, directly. So far these cover root Items and Subagents (their reads) and Session Artifacts (their reads, deletion and the staging of a Turn's export). `cmd/server` builds one `sessionpg.Store` and wires the Service and the Store into `execution.Dispatcher.Sessions` and `SessionsReader` and into the api fields; the Worker stages Artifacts through the Service. ## Request handling diff --git a/services/core/cmd/server/main.go b/services/core/cmd/server/main.go index 3466196eb..d3a39beaf 100644 --- a/services/core/cmd/server/main.go +++ b/services/core/cmd/server/main.go @@ -289,7 +289,7 @@ func run() error { } if registry != nil { dispatcher := &execution.Dispatcher{Store: executionStore, Registry: registry, - Credentials: vaultService, Observer: modelConfigurationStore, Deployment: deploymentService, + Credentials: vaultService, Observer: modelConfigurationStore, Deployment: deploymentService, DeploymentReader: deploymentStore, Sessions: sessionService, SessionsReader: sessionStore, ManagedRuntimes: managed, MaxConcurrentExecutions: concurrency} @@ -370,7 +370,7 @@ func run() error { } deps := api.Dependencies{ Engine: engine, Harnesses: kinds, CoreKeys: keyAdmin, - Installation: installation, InstallationBindings: executionStore, + Installation: installation, InstallationBindings: deploymentService, Projects: projectService, ProjectsReader: projectStore, ModelProviders: modelConfigurationService, ModelProvidersReader: modelConfigurationStore, Vaults: vaultService, VaultsReader: vaultStore, diff --git a/services/core/internal/api/errors.go b/services/core/internal/api/errors.go index 15e2ac270..13632f42b 100644 --- a/services/core/internal/api/errors.go +++ b/services/core/internal/api/errors.go @@ -86,16 +86,10 @@ func writeFieldError(w http.ResponseWriter, err error) bool { } // writeStoreError reports a failure of an operation that can meet a sandbox -// reset or a deployment's sandbox resources, then the Session errors. +// reset or the sandbox deployment, then the Session errors. func writeStoreError(w http.ResponseWriter, r *http.Request, err error) { - // A reset's resource check unwraps to deployment.ErrConflict and admission - // paused by a reset is Session admission's, so both precede the deployment errors. - var inUse *store.SandboxInUseError - switch { - case errors.As(err, &inUse): - writeCoreError(w, http.StatusConflict, "sandbox_in_use", "Hosted sandbox resources still belong to this deployment.", CoreErrorDetails{"allocations": CoreErrorNumber(float64(inUse.Resources.Allocations)), "pending": CoreErrorNumber(float64(inUse.Resources.Pending))}) - return - case errors.Is(err, store.ErrSandboxResetAdmission): + // Admission paused by a reset is Session admission's. + if errors.Is(err, store.ErrSandboxResetAdmission) { writeError(w, http.StatusServiceUnavailable, "sandbox_reset_in_progress", "A sandbox reset is in progress.") return } diff --git a/services/core/internal/api/errors_deployment.go b/services/core/internal/api/errors_deployment.go index 62b78dff7..873c9879c 100644 --- a/services/core/internal/api/errors_deployment.go +++ b/services/core/internal/api/errors_deployment.go @@ -48,7 +48,11 @@ func writeSandboxError(w http.ResponseWriter, err error) bool { var deploymentConfiguration *deployment.ConfigurationError var stale *deployment.GenerationStaleError var resetRequired *deployment.ResetRequiredError + var inUse *deployment.InUseError switch { + // Reset completion's resource check unwraps to deployment.ErrConflict. + case errors.As(err, &inUse): + writeCoreError(w, http.StatusConflict, "sandbox_in_use", "Hosted sandbox resources still belong to this deployment.", CoreErrorDetails{"allocations": CoreErrorNumber(float64(inUse.Resources.Allocations)), "pending": CoreErrorNumber(float64(inUse.Resources.Pending))}) case errors.As(err, &configuration): status := http.StatusInternalServerError switch configuration.Class { diff --git a/services/core/internal/api/fakes_test.go b/services/core/internal/api/fakes_test.go index d50f85999..2c6c71bcf 100644 --- a/services/core/internal/api/fakes_test.go +++ b/services/core/internal/api/fakes_test.go @@ -299,11 +299,11 @@ func (f *fakeDeploymentChanges) UpdateSandboxDeployment(a0 context.Context, a1 s type fakeDeploymentReset struct { t testing.TB - startSandboxReset func(context.Context, store.SandboxResetRequest) (deployment.View, error) + startSandboxReset func(context.Context, deployment.ResetRequest) (deployment.View, error) cancelSandboxReset func(context.Context, uint64) (deployment.View, error) } -func (f *fakeDeploymentReset) StartSandboxReset(a0 context.Context, a1 store.SandboxResetRequest) (deployment.View, error) { +func (f *fakeDeploymentReset) StartSandboxReset(a0 context.Context, a1 deployment.ResetRequest) (deployment.View, error) { if f.startSandboxReset == nil { unexpectedCall(f.t, "StartSandboxReset") } @@ -527,10 +527,10 @@ func (f *fakeInputAdmission) SubmitInputs(a0 context.Context, a1 string, a2 stri type fakeInstallationBindings struct { t testing.TB - addressBindings func(context.Context) (store.AddressBindings, error) + addressBindings func(context.Context) (deployment.AddressBindings, error) } -func (f *fakeInstallationBindings) AddressBindings(a0 context.Context) (store.AddressBindings, error) { +func (f *fakeInstallationBindings) AddressBindings(a0 context.Context) (deployment.AddressBindings, error) { if f.addressBindings == nil { unexpectedCall(f.t, "AddressBindings") } diff --git a/services/core/internal/api/installation.go b/services/core/internal/api/installation.go index e39cfe22d..d7328cec3 100644 --- a/services/core/internal/api/installation.go +++ b/services/core/internal/api/installation.go @@ -12,7 +12,7 @@ import ( "slices" "time" - "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/store" + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/deployment" ) // Installation reports Core's installation facts. Core reads them from its @@ -31,7 +31,7 @@ type Installation struct { SourceCommit *string `json:"source_commit" extensions:"x-nullable"` // The installer's settings snapshot (OAC_SETTINGS_FILE); null when the installer did not start Core. Configuration *InstallationConfiguration `json:"configuration" extensions:"x-nullable"` - AddressBindings store.AddressBindings `json:"address_bindings"` + AddressBindings deployment.AddressBindings `json:"address_bindings"` } // InstallationConfiguration says where process settings are changed and what @@ -99,7 +99,7 @@ func ParseInstallationConfiguration(raw []byte) (*InstallationConfiguration, err // InstallationBindings counts what is bound to the current public URL. type InstallationBindings interface { - AddressBindings(context.Context) (store.AddressBindings, error) + AddressBindings(context.Context) (deployment.AddressBindings, error) } // @Summary Retrieve installation facts and process settings @@ -113,7 +113,7 @@ type InstallationBindings interface { func (h *Handler) getInstallation(w http.ResponseWriter, r *http.Request) { bindings, err := h.InstallationBindings.AddressBindings(r.Context()) if err != nil { - writeStoreError(w, r, err) + writeDeploymentError(w, r, err) return } value := h.Installation diff --git a/services/core/internal/api/installation_test.go b/services/core/internal/api/installation_test.go index 37bc7f638..74dd81a6b 100644 --- a/services/core/internal/api/installation_test.go +++ b/services/core/internal/api/installation_test.go @@ -11,7 +11,6 @@ import ( "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/deployment" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sandbox" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sandbox/providers" - "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/store" ) func TestInstallationReadNeedsOnlyTheCoreKey(t *testing.T) { @@ -25,8 +24,8 @@ func TestInstallationReadNeedsOnlyTheCoreKey(t *testing.T) { if err != nil { t.Fatal(err) } - fakes.installationBindings.addressBindings = func(context.Context) (store.AddressBindings, error) { - return store.AddressBindings{Nodes: 2, NodesOnOtherAddress: 1}, nil + fakes.installationBindings.addressBindings = func(context.Context) (deployment.AddressBindings, error) { + return deployment.AddressBindings{Nodes: 2, NodesOnOtherAddress: 1}, nil } // No sandbox deployment: the read is available before any deployment exists. deps.Installation = Installation{InstallationID: &id, PublicURL: &public, Configuration: settings} diff --git a/services/core/internal/api/sandbox_deployment_changes_test.go b/services/core/internal/api/sandbox_deployment_changes_test.go index aee58af40..c83424794 100644 --- a/services/core/internal/api/sandbox_deployment_changes_test.go +++ b/services/core/internal/api/sandbox_deployment_changes_test.go @@ -24,7 +24,7 @@ func TestSandboxDeploymentChangesAuthenticateAndDecode(t *testing.T) { } return deployment.View{Provider: in.Provider}, nil } - maintain := func(_ context.Context, in store.SandboxResetRequest) (deployment.View, error) { + maintain := func(_ context.Context, in deployment.ResetRequest) (deployment.View, error) { resets++ if in.ExpectedGeneration != 2 { t.Fatal("generation was lost") @@ -73,36 +73,45 @@ func TestSandboxDeploymentChangesAuthenticateAndDecode(t *testing.T) { } } +// The deployment writer, which the reset handlers use, and the Session writer, +// which admission uses, report the deployment errors alike. func TestSandboxMutationErrorsExposeOnlyTypedCoreFacts(t *testing.T) { + writers := map[string]func(http.ResponseWriter, *http.Request, error){"deployment": writeDeploymentError, "store": writeStoreError} for _, tc := range []struct { - err error - code string - status int - details string + err error + code string + status int + details string + admission bool }{ - {&deployment.GenerationStaleError{CurrentGeneration: 8}, "sandbox_generation_stale", 409, `"current_generation":8`}, - {&deployment.ResetRequiredError{CurrentProvider: "docker", RequestedProvider: "e2b"}, "sandbox_reset_required", 409, `"requested_provider":"e2b"`}, - {&deployment.ResetRequiredError{CurrentProvider: "e2b", RequestedProvider: "e2b"}, "sandbox_reset_required", 409, `"current_provider":"e2b"`}, - {&store.SandboxInUseError{Resources: deployment.Resources{Allocations: 2, Pending: 1}}, "sandbox_in_use", 409, `"allocations":2`}, - {deployment.ErrResetInProgress, "sandbox_reset_in_progress", 409, ""}, - {store.ErrSandboxResetAdmission, "sandbox_reset_in_progress", 503, ""}, + {&deployment.GenerationStaleError{CurrentGeneration: 8}, "sandbox_generation_stale", 409, `"current_generation":8`, false}, + {&deployment.ResetRequiredError{CurrentProvider: "docker", RequestedProvider: "e2b"}, "sandbox_reset_required", 409, `"requested_provider":"e2b"`, false}, + {&deployment.ResetRequiredError{CurrentProvider: "e2b", RequestedProvider: "e2b"}, "sandbox_reset_required", 409, `"current_provider":"e2b"`, false}, + {&deployment.InUseError{Resources: deployment.Resources{Allocations: 2, Pending: 1}}, "sandbox_in_use", 409, `"allocations":2`, false}, + {deployment.ErrResetInProgress, "sandbox_reset_in_progress", 409, "", false}, + {store.ErrSandboxResetAdmission, "sandbox_reset_in_progress", 503, "", true}, } { - for _, core := range []bool{false, true} { - handler := http.Handler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { writeStoreError(w, r, tc.err) })) - if core { - handler = coreErrorResponses(handler) + for name, write := range writers { + if tc.admission && name != "store" { + continue } - response := httptest.NewRecorder() - handler.ServeHTTP(response, httptest.NewRequest("POST", "/test", nil)) - body := response.Body.String() - if response.Code != tc.status || !strings.Contains(body, `"code":"`+tc.code+`"`) { - t.Fatal(body) - } - if core && tc.details != "" && !strings.Contains(body, tc.details) { - t.Fatal("typed detail missing", body) - } - if !core && strings.Contains(body, `"details"`) { - t.Fatal("Core facts escaped their router", body) + for _, core := range []bool{false, true} { + handler := http.Handler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { write(w, r, tc.err) })) + if core { + handler = coreErrorResponses(handler) + } + response := httptest.NewRecorder() + handler.ServeHTTP(response, httptest.NewRequest("POST", "/test", nil)) + body := response.Body.String() + if response.Code != tc.status || !strings.Contains(body, `"code":"`+tc.code+`"`) { + t.Fatal(name, body) + } + if core && tc.details != "" && !strings.Contains(body, tc.details) { + t.Fatal("typed detail missing", body) + } + if !core && strings.Contains(body, `"details"`) { + t.Fatal("Core facts escaped their router", body) + } } } } diff --git a/services/core/internal/api/sandbox_deployment_setup.go b/services/core/internal/api/sandbox_deployment_setup.go index 74af0ad25..b58295306 100644 --- a/services/core/internal/api/sandbox_deployment_setup.go +++ b/services/core/internal/api/sandbox_deployment_setup.go @@ -3,14 +3,13 @@ package api import ( "context" "encoding/json" + "fmt" "net/http" "net/url" "strconv" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/deployment" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sandbox" - "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sessions" - "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/store" ) type SandboxDeploymentInput struct { @@ -47,7 +46,7 @@ type DeploymentChanges interface { // DeploymentReset starts and cancels a sandbox deployment reset through the // execution owner. type DeploymentReset interface { - StartSandboxReset(ctx context.Context, input store.SandboxResetRequest) (deployment.View, error) + StartSandboxReset(ctx context.Context, input deployment.ResetRequest) (deployment.View, error) CancelSandboxReset(ctx context.Context, generation uint64) (deployment.View, error) } @@ -123,7 +122,7 @@ func (h *Handler) updateSandboxDeployment(w http.ResponseWriter, r *http.Request // @Produce json // @Security DeploymentAdminAuth // @Accept json -// @Param body body store.SandboxResetRequest true "Reset mode and current deployment generation" +// @Param body body deployment.ResetRequest true "Reset mode and current deployment generation" // @Success 200 {object} deployment.View // @Failure 400,401,409,500,503 {object} CoreErrorResponse // @Router /core/v1/sandbox/deployment/reset [post] @@ -141,7 +140,7 @@ func (h *Handler) startSandboxReset(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "invalid_request_error", "A current expected_generation is required.", "expected_generation") return } - if input.Clear != "auto" && input.Clear != "force" { + if input.Clear != deployment.ResetAuto && input.Clear != deployment.ResetForce { writeError(w, http.StatusBadRequest, "invalid_request_error", "Choose auto or force for clear.", "clear") return } @@ -151,14 +150,15 @@ func (h *Handler) startSandboxReset(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "invalid_request_error", "deadline_seconds must be an integer when supplied.", "deadline_seconds") return } - if input.DeadlineSeconds != nil && (input.Clear != "auto" || *input.DeadlineSeconds < 300 || *input.DeadlineSeconds > 86400) { - writeCoreError(w, http.StatusBadRequest, "invalid_request_error", "deadline_seconds applies only to auto and must be between 300 and 86400.", CoreErrorDetails{"min": CoreErrorNumber(300), "max": CoreErrorNumber(86400)}, "deadline_seconds") + if input.DeadlineSeconds != nil && (input.Clear != deployment.ResetAuto || *input.DeadlineSeconds < deployment.MinResetDeadlineSeconds || *input.DeadlineSeconds > deployment.MaxResetDeadlineSeconds) { + message := fmt.Sprintf("deadline_seconds applies only to auto and must be between %d and %d.", deployment.MinResetDeadlineSeconds, deployment.MaxResetDeadlineSeconds) + writeCoreError(w, http.StatusBadRequest, "invalid_request_error", message, CoreErrorDetails{"min": CoreErrorNumber(deployment.MinResetDeadlineSeconds), "max": CoreErrorNumber(deployment.MaxResetDeadlineSeconds)}, "deadline_seconds") return } setAdminAuditSource(r, "") - result, err := h.Sandboxes.DeploymentReset.StartSandboxReset(r.Context(), store.SandboxResetRequest{ExpectedGeneration: *input.ExpectedGeneration, Clear: input.Clear, DeadlineSeconds: input.DeadlineSeconds}) + result, err := h.Sandboxes.DeploymentReset.StartSandboxReset(r.Context(), deployment.ResetRequest{ExpectedGeneration: *input.ExpectedGeneration, Clear: input.Clear, DeadlineSeconds: input.DeadlineSeconds}) if err != nil { - writeStoreError(w, r, err) + writeDeploymentError(w, r, err) return } writeJSON(w, http.StatusOK, result) @@ -182,7 +182,7 @@ func (h *Handler) cancelSandboxReset(w http.ResponseWriter, r *http.Request) { setAdminAuditSource(r, "") result, err := h.Sandboxes.DeploymentReset.CancelSandboxReset(r.Context(), query) if err != nil { - writeStoreError(w, r, err) + writeDeploymentError(w, r, err) return } writeJSON(w, http.StatusOK, result) @@ -190,7 +190,7 @@ func (h *Handler) cancelSandboxReset(w http.ResponseWriter, r *http.Request) { func parseResetGeneration(r *http.Request) (uint64, error) { query, err := url.ParseQuery(r.URL.RawQuery) if err != nil || len(query) != 1 || len(query["expected_generation"]) != 1 { - return 0, sessions.ErrInvalidInput + return 0, deployment.ErrInvalidInput } return strconv.ParseUint(query.Get("expected_generation"), 10, 64) } diff --git a/services/core/internal/deployment/errors.go b/services/core/internal/deployment/errors.go index 907c64226..3d6c6a1f8 100644 --- a/services/core/internal/deployment/errors.go +++ b/services/core/internal/deployment/errors.go @@ -49,6 +49,15 @@ func (e *ResetRequiredError) Error() string { } func (e *ResetRequiredError) Unwrap() error { return ErrConflict } +// InUseError rejects reset completion while hosted resources still belong to +// the deployment. +type InUseError struct{ Resources Resources } + +func (e *InUseError) Error() string { + return "hosted sandbox resources still belong to this deployment" +} +func (e *InUseError) Unwrap() error { return ErrConflict } + // ConfigurationError contains only validated, non-secret configuration // diagnostics. type ConfigurationError struct { diff --git a/services/core/internal/deployment/fakes_test.go b/services/core/internal/deployment/fakes_test.go index 97086a8d1..bdd5ee747 100644 --- a/services/core/internal/deployment/fakes_test.go +++ b/services/core/internal/deployment/fakes_test.go @@ -6,6 +6,7 @@ import ( "testing" "time" + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/adminaudit" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/coremetrics" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sandbox" ) @@ -67,15 +68,17 @@ func (f *fakeExecutionStorage) WithDeployment(ctx context.Context, apply func(De } type fakeReader struct { - t testing.TB - deployment func(context.Context) (Record, error) - snapshot func(context.Context) (Snapshot, error) - ownerEpoch func(context.Context) (uint64, error) - allocation func(context.Context, sandbox.Reference) (AllocationRecord, error) - generations func(context.Context, int64) ([]GenerationRecord, error) - nodes func(context.Context) ([]NodeRecord, error) - nodeHistory func(context.Context, string, coremetrics.Range) (NodeRecord, []HostHistoryPoint, error) - readNodes func(context.Context, func(NodeReads) error) error + t testing.TB + deployment func(context.Context) (Record, error) + snapshot func(context.Context) (Snapshot, error) + ownerEpoch func(context.Context) (uint64, error) + allocation func(context.Context, sandbox.Reference) (AllocationRecord, error) + generations func(context.Context, int64) ([]GenerationRecord, error) + nodes func(context.Context) ([]NodeRecord, error) + nodeHistory func(context.Context, string, coremetrics.Range) (NodeRecord, []HostHistoryPoint, error) + readNodes func(context.Context, func(NodeReads) error) error + resetSessions func(context.Context, string, bool) ([]ResetSession, error) + addressBindings func(context.Context, string) (AddressBindings, error) } func (f *fakeReader) Deployment(ctx context.Context) (Record, error) { @@ -134,6 +137,20 @@ func (f *fakeReader) ReadNodes(ctx context.Context, apply func(NodeReads) error) return f.readNodes(ctx, apply) } +func (f *fakeReader) ResetSessions(ctx context.Context, after string, force bool) ([]ResetSession, error) { + if f.resetSessions == nil { + unexpected(f.t, "ResetSessions") + } + return f.resetSessions(ctx, after, force) +} + +func (f *fakeReader) AddressBindings(ctx context.Context, publicURL string) (AddressBindings, error) { + if f.addressBindings == nil { + unexpected(f.t, "AddressBindings") + } + return f.addressBindings(ctx, publicURL) +} + type fakeNodeReads struct { t testing.TB loadDeployment func() (Record, error) @@ -325,7 +342,13 @@ type fakeDeploymentTx struct { recordConfigurationMetadata func(json.RawMessage) error retainGeneration func() error collectGenerations func() error + startReset func(string, int32, adminaudit.Source) error + forceReset func() error + cancelReset func() error + loadResetSource func() (adminaudit.Source, error) + completeReset func() error recordAudit func(string, string) error + recordAuditAs func(adminaudit.Source, string, string) error } func (f *fakeDeploymentTx) LoadDeployment() (Record, error) { @@ -425,3 +448,45 @@ func (f *fakeDeploymentTx) RecordAudit(action, installationID string) error { } return f.recordAudit(action, installationID) } + +func (f *fakeDeploymentTx) StartReset(clear string, deadlineSeconds int32, source adminaudit.Source) error { + if f.startReset == nil { + unexpected(f.t, "StartReset") + } + return f.startReset(clear, deadlineSeconds, source) +} + +func (f *fakeDeploymentTx) ForceReset() error { + if f.forceReset == nil { + unexpected(f.t, "ForceReset") + } + return f.forceReset() +} + +func (f *fakeDeploymentTx) CancelReset() error { + if f.cancelReset == nil { + unexpected(f.t, "CancelReset") + } + return f.cancelReset() +} + +func (f *fakeDeploymentTx) LoadResetSource() (adminaudit.Source, error) { + if f.loadResetSource == nil { + unexpected(f.t, "LoadResetSource") + } + return f.loadResetSource() +} + +func (f *fakeDeploymentTx) CompleteReset() error { + if f.completeReset == nil { + unexpected(f.t, "CompleteReset") + } + return f.completeReset() +} + +func (f *fakeDeploymentTx) RecordAuditAs(source adminaudit.Source, action, installationID string) error { + if f.recordAuditAs == nil { + unexpected(f.t, "RecordAuditAs") + } + return f.recordAuditAs(source, action, installationID) +} diff --git a/services/core/internal/deployment/public_url.go b/services/core/internal/deployment/public_url.go index 6d9dead16..bfa5f5178 100644 --- a/services/core/internal/deployment/public_url.go +++ b/services/core/internal/deployment/public_url.go @@ -1,6 +1,7 @@ package deployment import ( + "context" "net" "net/url" "strconv" @@ -60,3 +61,19 @@ func LoopbackOrigin(value string) bool { } return u.Hostname() == "localhost" } + +// AddressBindings counts what is bound to an installation address: nodes +// connect to the address they enrolled with, hosted sandboxes were started with +// the address current at the time, and self-hosted executors were installed +// with an advertised remote_url. +type AddressBindings struct { + Nodes int64 `json:"nodes"` + NodesOnOtherAddress int64 `json:"nodes_on_other_address"` + HostedSandboxes int64 `json:"hosted_sandboxes"` + SelfHostedExecutors int64 `json:"self_hosted_executors"` +} + +// AddressBindings counts what is bound to the installation public URL. +func (s *Service) AddressBindings(ctx context.Context) (AddressBindings, error) { + return s.reader.AddressBindings(ctx, s.publicURL) +} diff --git a/services/core/internal/deployment/reset.go b/services/core/internal/deployment/reset.go new file mode 100644 index 000000000..a404549b1 --- /dev/null +++ b/services/core/internal/deployment/reset.go @@ -0,0 +1,186 @@ +package deployment + +import ( + "context" + "math" + "time" + + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/adminaudit" +) + +// ResetRequest starts or escalates a reset of the sandbox deployment. +type ResetRequest struct { + ExpectedGeneration uint64 `json:"expected_generation" binding:"required" minimum:"0"` + Clear string `json:"clear" binding:"required" enums:"auto,force"` + DeadlineSeconds *int32 `json:"deadline_seconds,omitempty" minimum:"300" maximum:"86400"` +} + +// ResetSession is a hosted Session a reset still has to archive. +type ResetSession struct{ SessionID, TenantID string } + +const ( + // ResetAuto archives idle Sessions and escalates to ResetForce at the + // deadline. + ResetAuto = "auto" + // ResetForce archives every hosted Session, busy ones included. + ResetForce = "force" + + defaultResetDeadlineSeconds = 3600 +) + +// MinResetDeadlineSeconds and MaxResetDeadlineSeconds bound an auto reset's +// deadline. +const ( + MinResetDeadlineSeconds = 300 + MaxResetDeadlineSeconds = 86400 +) + +// resetDeadline validates the request's clear mode and returns the auto +// deadline in seconds. Only auto takes a deadline. +func resetDeadline(input ResetRequest) (int32, error) { + if input.Clear != ResetAuto && input.Clear != ResetForce { + return 0, ErrInvalidInput + } + if input.DeadlineSeconds == nil { + return defaultResetDeadlineSeconds, nil + } + if input.Clear != ResetAuto || *input.DeadlineSeconds < MinResetDeadlineSeconds || *input.DeadlineSeconds > MaxResetDeadlineSeconds { + return 0, ErrInvalidInput + } + return *input.DeadlineSeconds, nil +} + +// StartReset pauses admission and starts the reset, or escalates a running +// auto reset to force. Repeating the running mode is a no-op, and auto never +// downgrades force. The context must carry the administrator audit source, +// which the reset keeps for the audit entries it records later. It returns no +// view: the caller reads the deployment after commit. +func (e *ExecutionOperations) StartReset(ctx context.Context, installation string, input ResetRequest) error { + deadline, err := resetDeadline(input) + if err != nil { + return err + } + return e.storage.WithDeployment(ctx, func(tx DeploymentTx) error { + d, err := tx.LoadDeployment() + if err != nil { + return err + } + if err := checkGeneration(d, installation, input.ExpectedGeneration); err != nil { + return err + } + if d.Provider == "" { + return ErrNotConfigured + } + if d.Reset != nil { + if d.Reset.Clear == input.Clear { + return nil + } + if input.Clear != ResetForce { + return ErrResetInProgress + } + if err := tx.ForceReset(); err != nil { + return err + } + return tx.RecordAudit("reset_force", installation) + } + source, ok := adminaudit.FromContext(ctx) + if !ok { + return ErrInvalidInput + } + if err := tx.StartReset(input.Clear, deadline, source); err != nil { + return err + } + return tx.RecordAudit("reset_start", installation) + }) +} + +// CancelReset cancels a running reset and restores admission. Sessions it +// already archived stay archived. Without a running reset it only checks the +// generation. It returns no view: the caller reads the deployment after commit. +func (e *ExecutionOperations) CancelReset(ctx context.Context, installation string, generation uint64) error { + return e.storage.WithDeployment(ctx, func(tx DeploymentTx) error { + d, err := tx.LoadDeployment() + if err != nil { + return err + } + if err := checkGeneration(d, installation, generation); err != nil { + return err + } + if d.Reset == nil { + return nil + } + if err := tx.CancelReset(); err != nil { + return err + } + return tx.RecordAudit("reset_cancel", installation) + }) +} + +// AdvanceResetDeadline escalates an auto reset to force once its deadline has +// passed, before the reset selects work. The deadline is absolute, so a +// restart never extends it. The audit entry carries the administrator source +// that started the reset. +func (e *ExecutionOperations) AdvanceResetDeadline(ctx context.Context) error { + return e.storage.WithDeployment(ctx, func(tx DeploymentTx) error { + d, err := tx.LoadDeployment() + if err != nil { + return err + } + if d.Reset == nil || d.Reset.Clear != ResetAuto || d.Reset.DeadlineAt == nil || time.Now().Before(*d.Reset.DeadlineAt) { + return nil + } + source, err := tx.LoadResetSource() + if err != nil { + return err + } + if err := tx.ForceReset(); err != nil { + return err + } + return tx.RecordAuditAs(source, "reset_deadline", d.InstallationID) + }) +} + +// CompleteReset clears the drained deployment: it forgets the selection, +// retires the nodes, enrollments and retained generations, and advances the +// generation and owner epoch by one. It runs after the manager has drained +// and takes no Session lock, because archive and admission lock the Session +// before the deployment. It returns the committed, unconfigured generation, +// and InUseError while hosted resources remain. +func (e *ExecutionOperations) CompleteReset(ctx context.Context, installation string, generation uint64, requestedAt time.Time) (uint64, error) { + var committed uint64 + err := e.storage.WithDeployment(ctx, func(tx DeploymentTx) error { + d, err := tx.LoadDeployment() + if err != nil { + return err + } + if err := checkGeneration(d, installation, generation); err != nil { + return err + } + if d.Reset == nil || !d.Reset.RequestedAt.Equal(requestedAt) { + return ErrConflict + } + if d.Generation >= math.MaxInt64 || d.OwnerEpoch >= math.MaxInt64 { + return ErrConflict + } + resources, err := tx.CountResources() + if err != nil { + return err + } + if resources.Allocations != 0 || resources.Pending != 0 { + return &InUseError{Resources: resources} + } + source, err := tx.LoadResetSource() + if err != nil { + return err + } + if err := tx.CompleteReset(); err != nil { + return err + } + committed = d.Generation + 1 + return tx.RecordAuditAs(source, "reset_complete", installation) + }) + if err != nil { + return 0, err + } + return committed, nil +} diff --git a/services/core/internal/deployment/reset_test.go b/services/core/internal/deployment/reset_test.go new file mode 100644 index 000000000..1a2c1187a --- /dev/null +++ b/services/core/internal/deployment/reset_test.go @@ -0,0 +1,161 @@ +package deployment + +import ( + "context" + "errors" + "fmt" + "math" + "slices" + "testing" + "time" + + "github.com/google/uuid" + + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/adminaudit" +) + +// resetTx serves stored and records each reset write, with its arguments. +func resetTx(t *testing.T, stored Record, resources Resources, source adminaudit.Source, calls *[]string) *fakeDeploymentTx { + record := func(format string, args ...any) error { + *calls = append(*calls, fmt.Sprintf(format, args...)) + return nil + } + return &fakeDeploymentTx{t: t, + loadDeployment: func() (Record, error) { return stored, nil }, + countResources: func() (Resources, error) { return resources, nil }, + loadResetSource: func() (adminaudit.Source, error) { return source, nil }, + startReset: func(clear string, deadline int32, source adminaudit.Source) error { + return record("StartReset %s %d %s", clear, deadline, source.CredentialID) + }, + forceReset: func() error { return record("ForceReset") }, + cancelReset: func() error { return record("CancelReset") }, + completeReset: func() error { return record("CompleteReset") }, + recordAudit: func(action, installation string) error { return record("RecordAudit %s %s", action, installation) }, + recordAuditAs: func(source adminaudit.Source, action, installation string) error { + return record("RecordAuditAs %s %s %s", source.CredentialID, action, installation) + }, + } +} + +func resetting(t *testing.T, installation, clear string, requestedAt time.Time, deadline *time.Time) Record { + d := webDeployment(t, installation, "docker", 4) + d.Reset = &ResetState{Clear: clear, RequestedAt: requestedAt, DeadlineAt: deadline} + return d +} + +func TestStartResetDecisions(t *testing.T) { + installation := uuid.NewString() + admin := adminaudit.WithSource(t.Context(), adminaudit.Source{CredentialID: "admin"}) + idle := webDeployment(t, installation, "docker", 4) + unconfigured := idle + unconfigured.Provider = "" + seconds := func(v int32) *int32 { return &v } + for name, test := range map[string]struct { + ctx context.Context + stored Record + input ResetRequest + want error + calls []string + }{ + "unknown clear": {admin, idle, ResetRequest{ExpectedGeneration: 4, Clear: "sandboxes"}, ErrInvalidInput, nil}, + "deadline with force": {admin, idle, ResetRequest{ExpectedGeneration: 4, Clear: ResetForce, DeadlineSeconds: seconds(600)}, ErrInvalidInput, nil}, + "deadline below minimum": {admin, idle, ResetRequest{ExpectedGeneration: 4, Clear: ResetAuto, DeadlineSeconds: seconds(299)}, ErrInvalidInput, nil}, + "deadline above maximum": {admin, idle, ResetRequest{ExpectedGeneration: 4, Clear: ResetAuto, DeadlineSeconds: seconds(86401)}, ErrInvalidInput, nil}, + "stale generation": {admin, idle, ResetRequest{ExpectedGeneration: 3, Clear: ResetAuto}, ErrConflict, nil}, + "not configured": {admin, unconfigured, ResetRequest{ExpectedGeneration: 4, Clear: ResetAuto}, ErrNotConfigured, nil}, + "no administrator source": {t.Context(), idle, ResetRequest{ExpectedGeneration: 4, Clear: ResetAuto}, ErrInvalidInput, nil}, + "auto with default": {admin, idle, ResetRequest{ExpectedGeneration: 4, Clear: ResetAuto}, nil, []string{"StartReset auto 3600 admin", "RecordAudit reset_start " + installation}}, + "auto with deadline": {admin, idle, ResetRequest{ExpectedGeneration: 4, Clear: ResetAuto, DeadlineSeconds: seconds(86400)}, nil, []string{"StartReset auto 86400 admin", "RecordAudit reset_start " + installation}}, + "force": {admin, idle, ResetRequest{ExpectedGeneration: 4, Clear: ResetForce}, nil, []string{"StartReset force 3600 admin", "RecordAudit reset_start " + installation}}, + "same mode is a no-op": {admin, resetting(t, installation, ResetAuto, time.Now(), nil), ResetRequest{ExpectedGeneration: 4, Clear: ResetAuto}, nil, nil}, + "force escalates auto": {admin, resetting(t, installation, ResetAuto, time.Now(), nil), ResetRequest{ExpectedGeneration: 4, Clear: ResetForce}, nil, []string{"ForceReset", "RecordAudit reset_force " + installation}}, + "auto never replaces force": {admin, resetting(t, installation, ResetForce, time.Now(), nil), ResetRequest{ExpectedGeneration: 4, Clear: ResetAuto}, ErrResetInProgress, nil}, + } { + var calls []string + err := operations(t, testPublicURL, resetTx(t, test.stored, Resources{}, adminaudit.Source{}, &calls)).StartReset(test.ctx, installation, test.input) + if !errors.Is(err, test.want) || (test.want == nil && err != nil) || !slices.Equal(calls, test.calls) { + t.Errorf("%s: StartReset = %v with %q, want %v with %q", name, err, calls, test.want, test.calls) + } + } +} + +func TestCancelResetDecisions(t *testing.T) { + installation := uuid.NewString() + for name, test := range map[string]struct { + stored Record + generation uint64 + want error + calls []string + }{ + "stale generation": {resetting(t, installation, ResetAuto, time.Now(), nil), 3, ErrConflict, nil}, + "no running reset": {webDeployment(t, installation, "docker", 4), 4, nil, nil}, + "cancels the reset": {resetting(t, installation, ResetAuto, time.Now(), nil), 4, nil, []string{"CancelReset", "RecordAudit reset_cancel " + installation}}, + } { + var calls []string + err := operations(t, testPublicURL, resetTx(t, test.stored, Resources{}, adminaudit.Source{}, &calls)).CancelReset(t.Context(), installation, test.generation) + if !errors.Is(err, test.want) || (test.want == nil && err != nil) || !slices.Equal(calls, test.calls) { + t.Errorf("%s: CancelReset = %v with %q, want %v with %q", name, err, calls, test.want, test.calls) + } + } +} + +// A passed auto deadline escalates to force under the source that started the +// reset, never the caller's. +func TestAdvanceResetDeadlineUsesTheStoredSource(t *testing.T) { + installation := uuid.NewString() + past, future := time.Now().Add(-time.Second), time.Now().Add(time.Hour) + starter := adminaudit.Source{CredentialID: "starter"} + for name, test := range map[string]struct { + stored Record + calls []string + }{ + "no reset": {webDeployment(t, installation, "docker", 4), nil}, + "deadline ahead": {resetting(t, installation, ResetAuto, time.Now(), &future), nil}, + "already force": {resetting(t, installation, ResetForce, time.Now(), nil), nil}, + "deadline passed": {resetting(t, installation, ResetAuto, time.Now(), &past), + []string{"ForceReset", "RecordAuditAs starter reset_deadline " + installation}}, + } { + var calls []string + if err := operations(t, testPublicURL, resetTx(t, test.stored, Resources{}, starter, &calls)).AdvanceResetDeadline(t.Context()); err != nil || !slices.Equal(calls, test.calls) { + t.Errorf("%s: AdvanceResetDeadline = %v with %q, want %q", name, err, calls, test.calls) + } + } +} + +func TestCompleteResetDecisions(t *testing.T) { + installation := uuid.NewString() + requestedAt := time.Now() + starter := adminaudit.Source{CredentialID: "starter"} + running := resetting(t, installation, ResetForce, requestedAt, nil) + exhausted := running + exhausted.OwnerEpoch = math.MaxInt64 + for name, test := range map[string]struct { + stored Record + generation uint64 + requestedAt time.Time + resources Resources + }{ + "stale generation": {running, 3, requestedAt, Resources{}}, + "no running reset": {webDeployment(t, installation, "docker", 4), 4, requestedAt, Resources{}}, + "another reset": {running, 4, requestedAt.Add(time.Second), Resources{}}, + "owner epoch exhausted": {exhausted, 4, requestedAt, Resources{}}, + "resources still allocated": {running, 4, requestedAt, Resources{Allocations: 2, Pending: 1}}, + } { + var calls []string + if _, err := operations(t, testPublicURL, resetTx(t, test.stored, test.resources, starter, &calls)).CompleteReset(t.Context(), installation, test.generation, test.requestedAt); !errors.Is(err, ErrConflict) || len(calls) != 0 { + t.Errorf("%s: CompleteReset = %v with %q, want a conflict and no writes", name, err, calls) + } + } + + var calls []string + var inUse *InUseError + _, err := operations(t, testPublicURL, resetTx(t, running, Resources{Allocations: 2, Pending: 1}, starter, &calls)).CompleteReset(t.Context(), installation, 4, requestedAt) + if !errors.As(err, &inUse) || inUse.Resources != (Resources{Allocations: 2, Pending: 1}) { + t.Fatalf("CompleteReset with resources = %v", err) + } + + committed, err := operations(t, testPublicURL, resetTx(t, running, Resources{}, starter, &calls)).CompleteReset(t.Context(), installation, 4, requestedAt) + if want := []string{"CompleteReset", "RecordAuditAs starter reset_complete " + installation}; err != nil || committed != 5 || !slices.Equal(calls, want) { + t.Fatalf("CompleteReset = %d, %v with %q, want 5 with %q", committed, err, calls, want) + } +} diff --git a/services/core/internal/deployment/storage.go b/services/core/internal/deployment/storage.go index 5f09f0992..05f41e34f 100644 --- a/services/core/internal/deployment/storage.go +++ b/services/core/internal/deployment/storage.go @@ -5,6 +5,7 @@ import ( "encoding/json" "time" + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/adminaudit" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/coremetrics" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sandbox" ) @@ -63,6 +64,14 @@ type Reader interface { NodeHistory(ctx context.Context, nodeID string, window coremetrics.Range) (NodeRecord, []HostHistoryPoint, error) // ReadNodes runs apply in one read-only snapshot. ReadNodes(ctx context.Context, apply func(NodeReads) error) error + // ResetSessions returns up to 32 hosted Sessions after the given Session + // ID, in ID order, that still hold or await deployment resources. Without + // force it skips Sessions with a running Turn or a pending file write. An + // empty after starts at the beginning; a malformed one is ErrInvalidInput. + ResetSessions(ctx context.Context, after string, force bool) ([]ResetSession, error) + // AddressBindings counts what is bound to an installation address, read + // in one snapshot. publicURL is the address nodes are compared against. + AddressBindings(ctx context.Context, publicURL string) (AddressBindings, error) } // NodeReads loads node authentication facts. @@ -131,8 +140,27 @@ type DeploymentTx interface { RetainGeneration() error // CollectGenerations deletes retained generations nothing uses. CollectGenerations() error - // RecordAudit records an administrator mutation of the deployment. + // StartReset pauses admission and records a reset in the clear mode, + // with its request time, the auto deadline in seconds from now and the + // administrator source that started it. + StartReset(clear string, deadlineSeconds int32, source adminaudit.Source) error + // ForceReset escalates a running auto reset to force. + ForceReset() error + // CancelReset forgets the running reset and restores admission. + CancelReset() error + // LoadResetSource returns the administrator source that started the + // running reset. + LoadResetSource() (adminaudit.Source, error) + // CompleteReset forgets the selection and the reset, restores admission, + // advances the generation and owner epoch by one, and retires the nodes, + // enrollments and retained generations. + CompleteReset() error + // RecordAudit records an administrator mutation of the deployment by the + // source the transaction's context carries. RecordAudit(action, installationID string) error + // RecordAuditAs records an administrator mutation of the deployment by + // the given source. + RecordAuditAs(source adminaudit.Source, action, installationID string) error } // Record is the stored deployment. diff --git a/services/core/internal/execution/archive_cancellation_cleanup_test.go b/services/core/internal/execution/archive_cancellation_cleanup_test.go index 8c7a2f739..709cefe8e 100644 --- a/services/core/internal/execution/archive_cancellation_cleanup_test.go +++ b/services/core/internal/execution/archive_cancellation_cleanup_test.go @@ -64,7 +64,7 @@ func TestArchiveWaitingCleanupReceiptBarrier(t *testing.T) { }{{"Kill_no_delivery", false, false}, {"KillCompute_no_delivery", true, false}, {"Kill_live_delivery", false, true}, {"KillCompute_live_delivery", true, true}} { t.Run(scenario.name, func(t *testing.T) { checkpoint := scenario.checkpoint - s, leased, _, pool := resetManagerStoreDB(t, nil) + s, leased, _, _, pool := resetManagerStoreDB(t, nil) writer := leased.Store installation := initializeE2BDeployment(t, leased) projectID := uuid.NewString() diff --git a/services/core/internal/execution/deployment_fixture_test.go b/services/core/internal/execution/deployment_fixture_test.go index 3ec740d90..a1c794fa6 100644 --- a/services/core/internal/execution/deployment_fixture_test.go +++ b/services/core/internal/execution/deployment_fixture_test.go @@ -20,13 +20,14 @@ import ( // a public origin, so it is never loopback. const fixturePublicURL = "https://core.example" -// testDeployment builds the pooled deployment service and the deployment -// execution operations on lease, as cmd/server does for the Worker. cipher is -// nil when the owner has no credential key. -func testDeployment(t *testing.T, pool *pgxpool.Pool, cipher *credentialcrypto.Cipher, lease *pgunit.Lease) (*deployment.Service, *deployment.ExecutionOperations) { +// testDeployment builds the pooled deployment service and reader and the +// deployment execution operations on lease, as cmd/server does for the Worker. +// cipher is nil when the owner has no credential key. +func testDeployment(t *testing.T, pool *pgxpool.Pool, cipher *credentialcrypto.Cipher, lease *pgunit.Lease) (*deployment.Service, deployment.Reader, *deployment.ExecutionOperations) { t.Helper() adapter := deploymentpg.New(pgunit.NewPool(pool), cipher) - return deploymentOperations(t, adapter, adapter, deploymentpg.NewExecution(lease, cipher)) + service, operations := deploymentOperations(t, adapter, adapter, deploymentpg.NewExecution(lease, cipher)) + return service, adapter, operations } // unitDeploymentService builds a deployment service for tests without a @@ -113,15 +114,17 @@ func (s *strictExecutionStorage) WithDeployment(ctx context.Context, apply func( // strictDeploymentReader runs each set func; any other call fails the test. type strictDeploymentReader struct { - t *testing.T - deployment func(context.Context) (deployment.Record, error) - snapshot func(context.Context) (deployment.Snapshot, error) - ownerEpoch func(context.Context) (uint64, error) - allocation func(context.Context, sandbox.Reference) (deployment.AllocationRecord, error) - generations func(context.Context, int64) ([]deployment.GenerationRecord, error) - nodes func(context.Context) ([]deployment.NodeRecord, error) - nodeHistory func(context.Context, string, coremetrics.Range) (deployment.NodeRecord, []deployment.HostHistoryPoint, error) - readNodes func(context.Context, func(deployment.NodeReads) error) error + t *testing.T + deployment func(context.Context) (deployment.Record, error) + snapshot func(context.Context) (deployment.Snapshot, error) + ownerEpoch func(context.Context) (uint64, error) + allocation func(context.Context, sandbox.Reference) (deployment.AllocationRecord, error) + generations func(context.Context, int64) ([]deployment.GenerationRecord, error) + nodes func(context.Context) ([]deployment.NodeRecord, error) + nodeHistory func(context.Context, string, coremetrics.Range) (deployment.NodeRecord, []deployment.HostHistoryPoint, error) + readNodes func(context.Context, func(deployment.NodeReads) error) error + resetSessions func(context.Context, string, bool) ([]deployment.ResetSession, error) + addressBindings func(context.Context, string) (deployment.AddressBindings, error) } func (r *strictDeploymentReader) Deployment(ctx context.Context) (deployment.Record, error) { @@ -179,3 +182,17 @@ func (r *strictDeploymentReader) ReadNodes(ctx context.Context, apply func(deplo } return r.readNodes(ctx, apply) } + +func (r *strictDeploymentReader) ResetSessions(ctx context.Context, after string, force bool) ([]deployment.ResetSession, error) { + if r.resetSessions == nil { + return nil, unexpectedDeploymentCall(r.t, "ResetSessions") + } + return r.resetSessions(ctx, after, force) +} + +func (r *strictDeploymentReader) AddressBindings(ctx context.Context, publicURL string) (deployment.AddressBindings, error) { + if r.addressBindings == nil { + return deployment.AddressBindings{}, unexpectedDeploymentCall(r.t, "AddressBindings") + } + return r.addressBindings(ctx, publicURL) +} diff --git a/services/core/internal/execution/deployment_provider_observations_test.go b/services/core/internal/execution/deployment_provider_observations_test.go index 845cdc7ef..80c6c50df 100644 --- a/services/core/internal/execution/deployment_provider_observations_test.go +++ b/services/core/internal/execution/deployment_provider_observations_test.go @@ -36,7 +36,7 @@ type finishObservationFixture struct { func newFinishObservationFixture(t *testing.T, maxConnections int32) finishObservationFixture { t.Helper() var cfg *pgxpool.Config - s, owner, _ := resetManagerStoreConfig(t, func(c *pgxpool.Config) { + s, owner, _, _ := resetManagerStoreConfig(t, func(c *pgxpool.Config) { if maxConnections > 0 { c.MaxConns = maxConnections } diff --git a/services/core/internal/execution/dispatcher.go b/services/core/internal/execution/dispatcher.go index 7dd709059..fb99e4556 100644 --- a/services/core/internal/execution/dispatcher.go +++ b/services/core/internal/execution/dispatcher.go @@ -40,6 +40,9 @@ type Dispatcher struct { // Deployment reads the sandbox deployment and prepares a selection's setup. // It is required; deployment changes go through Owner.Deployment. Deployment *deployment.Service + // DeploymentReader reads the deployment's pooled records, such as the + // Sessions a reset still has to archive. It is required. + DeploymentReader deployment.Reader // Sessions runs the pooled Session use cases, such as staging a Turn's // Artifacts. It is required. Sessions *sessions.Service diff --git a/services/core/internal/execution/owner_test.go b/services/core/internal/execution/owner_test.go index 5dae77e5c..441ff57da 100644 --- a/services/core/internal/execution/owner_test.go +++ b/services/core/internal/execution/owner_test.go @@ -101,35 +101,40 @@ func TestStartWorkerFailureClosesLeaseOnce(t *testing.T) { _, err := StartWorker(canceled, &Dispatcher{Credentials: &recordingCredentials{}, Observer: unusedObserver{t}}, Owner{Lease: lease}) return err }, - "missing Session service": func(t *testing.T, lease *closeCountingLease) error { - _, _, deployments := resetManagerStore(t) + "missing deployment reader": func(t *testing.T, lease *closeCountingLease) error { + _, _, deployments, _ := resetManagerStore(t) _, err := StartWorker(canceled, &Dispatcher{Credentials: &recordingCredentials{}, Observer: unusedObserver{t}, Deployment: deployments}, Owner{Lease: lease}) return err }, + "missing Session service": func(t *testing.T, lease *closeCountingLease) error { + _, _, deployments, deploymentReader := resetManagerStore(t) + _, err := StartWorker(canceled, &Dispatcher{Credentials: &recordingCredentials{}, Observer: unusedObserver{t}, Deployment: deployments, DeploymentReader: deploymentReader}, Owner{Lease: lease}) + return err + }, "missing Session reader": func(t *testing.T, lease *closeCountingLease) error { - _, _, deployments := resetManagerStore(t) + _, _, deployments, deploymentReader := resetManagerStore(t) service, _ := unusedSessions(t) - _, err := StartWorker(canceled, &Dispatcher{Credentials: &recordingCredentials{}, Observer: unusedObserver{t}, Deployment: deployments, Sessions: service}, Owner{Lease: lease}) + _, err := StartWorker(canceled, &Dispatcher{Credentials: &recordingCredentials{}, Observer: unusedObserver{t}, Deployment: deployments, DeploymentReader: deploymentReader, Sessions: service}, Owner{Lease: lease}) return err }, "missing Store": func(t *testing.T, lease *closeCountingLease) error { - _, _, deployments := resetManagerStore(t) + _, _, deployments, deploymentReader := resetManagerStore(t) service, reader := unusedSessions(t) - _, err := StartWorker(canceled, &Dispatcher{Credentials: &recordingCredentials{}, Observer: unusedObserver{t}, Deployment: deployments, Sessions: service, SessionsReader: reader}, Owner{Lease: lease}) + _, err := StartWorker(canceled, &Dispatcher{Credentials: &recordingCredentials{}, Observer: unusedObserver{t}, Deployment: deployments, DeploymentReader: deploymentReader, Sessions: service, SessionsReader: reader}, Owner{Lease: lease}) return err }, "missing deployment": func(t *testing.T, lease *closeCountingLease) error { - _, owner, deployments := resetManagerStore(t) + _, owner, deployments, deploymentReader := resetManagerStore(t) service, reader := unusedSessions(t) - _, err := StartWorker(canceled, &Dispatcher{Credentials: &recordingCredentials{}, Observer: unusedObserver{t}, Deployment: deployments, Sessions: service, SessionsReader: reader}, Owner{Lease: lease, Store: owner.Store}) + _, err := StartWorker(canceled, &Dispatcher{Credentials: &recordingCredentials{}, Observer: unusedObserver{t}, Deployment: deployments, DeploymentReader: deploymentReader, Sessions: service, SessionsReader: reader}, Owner{Lease: lease, Store: owner.Store}) return err }, "deployment claim": func(t *testing.T, lease *closeCountingLease) error { - s, owner, deployments := resetManagerStore(t) + s, owner, deployments, deploymentReader := resetManagerStore(t) lease.inner = owner.Lease id := uuid.NewString() service, reader := unusedSessions(t) - dispatcher := &Dispatcher{Store: s, Registry: runtimegateway.NewRegistry(), Credentials: &recordingCredentials{}, Observer: unusedObserver{t}, Deployment: deployments, Sessions: service, SessionsReader: reader, ManagedRuntimes: NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return nil, nil })} + dispatcher := &Dispatcher{Store: s, Registry: runtimegateway.NewRegistry(), Credentials: &recordingCredentials{}, Observer: unusedObserver{t}, Deployment: deployments, DeploymentReader: deploymentReader, Sessions: service, SessionsReader: reader, ManagedRuntimes: NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return nil, nil })} _, err := StartWorker(canceled, dispatcher, Owner{Lease: lease, Store: owner.Store, Deployment: owner.Deployment}) if ping := owner.Lease.CheckOwnership(t.Context()); !errors.Is(ping, pgunit.ErrLeaseClosed) { t.Error("failed start kept the database lease", ping) @@ -151,7 +156,7 @@ func TestStartWorkerFailureClosesLeaseOnce(t *testing.T) { } func TestStartWorkerChecksDeploymentAfterItsDependencies(t *testing.T) { - _, owner, deployments := resetManagerStore(t) + _, owner, deployments, deploymentReader := resetManagerStore(t) credentials, observer := &recordingCredentials{}, unusedObserver{t} service, reader := unusedSessions(t) for _, test := range []struct { @@ -163,10 +168,11 @@ func TestStartWorkerChecksDeploymentAfterItsDependencies(t *testing.T) { {"missing Credentials", Dispatcher{Observer: observer}, true, "execution worker requires MCP Credentials"}, {"missing observer", Dispatcher{Credentials: credentials}, true, "execution requires a model configuration observer"}, {"missing deployment service", Dispatcher{Credentials: credentials, Observer: observer}, true, "execution worker requires the deployment service"}, - {"missing Session service", Dispatcher{Credentials: credentials, Observer: observer, Deployment: deployments}, true, "execution worker requires the Session service"}, - {"missing Session reader", Dispatcher{Credentials: credentials, Observer: observer, Deployment: deployments, Sessions: service}, true, "execution worker requires the Session reader"}, - {"missing Store", Dispatcher{Credentials: credentials, Observer: observer, Deployment: deployments, Sessions: service, SessionsReader: reader}, false, "execution worker requires the execution Store"}, - {"missing deployment", Dispatcher{Credentials: credentials, Observer: observer, Deployment: deployments, Sessions: service, SessionsReader: reader}, true, "execution worker requires the deployment execution operations"}, + {"missing deployment reader", Dispatcher{Credentials: credentials, Observer: observer, Deployment: deployments}, true, "execution worker requires the deployment reader"}, + {"missing Session service", Dispatcher{Credentials: credentials, Observer: observer, Deployment: deployments, DeploymentReader: deploymentReader}, true, "execution worker requires the Session service"}, + {"missing Session reader", Dispatcher{Credentials: credentials, Observer: observer, Deployment: deployments, DeploymentReader: deploymentReader, Sessions: service}, true, "execution worker requires the Session reader"}, + {"missing Store", Dispatcher{Credentials: credentials, Observer: observer, Deployment: deployments, DeploymentReader: deploymentReader, Sessions: service, SessionsReader: reader}, false, "execution worker requires the execution Store"}, + {"missing deployment", Dispatcher{Credentials: credentials, Observer: observer, Deployment: deployments, DeploymentReader: deploymentReader, Sessions: service, SessionsReader: reader}, true, "execution worker requires the deployment execution operations"}, } { t.Run(test.name, func(t *testing.T) { lease := &closeCountingLease{t: t} @@ -183,11 +189,11 @@ func TestStartWorkerChecksDeploymentAfterItsDependencies(t *testing.T) { } func TestWorkerRunClosesLeaseAfterDrain(t *testing.T) { - s, owner, deployments := resetManagerStore(t) + s, owner, deployments, deploymentReader := resetManagerStore(t) lease := &closeCountingLease{t: t, inner: owner.Lease} id := uuid.NewString() service, reader := unusedSessions(t) - dispatcher := &Dispatcher{Store: s, Registry: runtimegateway.NewRegistry(), Credentials: &recordingCredentials{}, Observer: unusedObserver{t}, Deployment: deployments, Sessions: service, SessionsReader: reader, ManagedRuntimes: NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return nil, nil })} + dispatcher := &Dispatcher{Store: s, Registry: runtimegateway.NewRegistry(), Credentials: &recordingCredentials{}, Observer: unusedObserver{t}, Deployment: deployments, DeploymentReader: deploymentReader, Sessions: service, SessionsReader: reader, ManagedRuntimes: NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return nil, nil })} worker, err := StartWorker(t.Context(), dispatcher, Owner{Lease: lease, Store: owner.Store, Deployment: owner.Deployment}) if err != nil { t.Fatal(err) diff --git a/services/core/internal/execution/runtime_lifecycle.go b/services/core/internal/execution/runtime_lifecycle.go index 981506c6d..dafa48f45 100644 --- a/services/core/internal/execution/runtime_lifecycle.go +++ b/services/core/internal/execution/runtime_lifecycle.go @@ -58,7 +58,7 @@ type runtimeLifecycle struct { wakeHints chan struct{} } -func newRuntimeManager(owner Owner, deployments *deployment.Service, registry *runtimegateway.Registry, config *RuntimeProvider) (*runtimeManager, error) { +func newRuntimeManager(owner Owner, deployments *deployment.Service, reader deployment.Reader, registry *runtimegateway.Registry, config *RuntimeProvider) (*runtimeManager, error) { if config == nil { return nil, nil } @@ -76,7 +76,7 @@ func newRuntimeManager(owner Owner, deployments *deployment.Service, registry *r } } ctx, stop := context.WithCancel(context.Background()) - return &runtimeManager{store: owner.Store, deployment: owner.Deployment, deploymentService: deployments, lease: owner.Lease, registry: registry, config: copied, setupInstallationID: config.InstallationID, loadDeployment: config.loadDeployment, prepareDeployment: config.prepareDeployment, publishUnconfigured: config.PublishUnconfigured, setupGate: make(chan struct{}, 1), mutationGate: make(chan struct{}, 1), ctx: ctx, cancel: stop, nodes: make(map[string]*runtimeNode), failed: make(chan error, 1), inventory: make(chan struct{}, 1)}, nil + return &runtimeManager{store: owner.Store, deployment: owner.Deployment, deploymentService: deployments, deploymentReader: reader, lease: owner.Lease, registry: registry, config: copied, setupInstallationID: config.InstallationID, loadDeployment: config.loadDeployment, prepareDeployment: config.prepareDeployment, publishUnconfigured: config.PublishUnconfigured, setupGate: make(chan struct{}, 1), mutationGate: make(chan struct{}, 1), ctx: ctx, cancel: stop, nodes: make(map[string]*runtimeNode), failed: make(chan error, 1), inventory: make(chan struct{}, 1)}, nil } func validatedRuntimeProvider(config *RuntimeProvider, registry *runtimegateway.Registry) (RuntimeProvider, error) { diff --git a/services/core/internal/execution/runtime_manager.go b/services/core/internal/execution/runtime_manager.go index 4ace4764f..95b9acedf 100644 --- a/services/core/internal/execution/runtime_manager.go +++ b/services/core/internal/execution/runtime_manager.go @@ -20,6 +20,7 @@ type runtimeManager struct { store *store.Store deployment *deployment.ExecutionOperations deploymentService *deployment.Service + deploymentReader deployment.Reader lease Ownership registry *runtimegateway.Registry config RuntimeProvider diff --git a/services/core/internal/execution/runtime_retirement_failure_test.go b/services/core/internal/execution/runtime_retirement_failure_test.go index c0065b04e..6a004fa75 100644 --- a/services/core/internal/execution/runtime_retirement_failure_test.go +++ b/services/core/internal/execution/runtime_retirement_failure_test.go @@ -29,7 +29,7 @@ func TestFailedInventoryRetirementClosesAdmissionAndRetainsGate(t *testing.T) { var readOnce sync.Once unblockRead := func() { readOnce.Do(func() { close(releaseRead) }) } defer unblockRead() - owner, _, pool := delayedReadWriter(t, &armed, reading, releaseRead) + owner, _, _, pool := delayedReadWriter(t, &armed, reading, releaseRead) m := testRuntimeManager(t) m.store, m.lease = owner.Store, owner.Lease m.loadDeployment = func(context.Context) (*RuntimeProvider, error) { return nil, nil } diff --git a/services/core/internal/execution/sandbox_deployment_drain_test.go b/services/core/internal/execution/sandbox_deployment_drain_test.go index ccfc8b937..15467107b 100644 --- a/services/core/internal/execution/sandbox_deployment_drain_test.go +++ b/services/core/internal/execution/sandbox_deployment_drain_test.go @@ -48,7 +48,7 @@ func (c *delayedLeaseRead) Read(p []byte) (int, error) { // delayedReadWriter owns an isolated database, so each case has its own // advisory-lock namespace. The delayed driver read lets a cancellation fence hit // its own deadline without shortening production timeouts. -func delayedReadWriter(t *testing.T, armed *atomic.Bool, reading chan struct{}, release <-chan struct{}) (Owner, *deployment.Service, *pgxpool.Pool) { +func delayedReadWriter(t *testing.T, armed *atomic.Bool, reading chan struct{}, release <-chan struct{}) (Owner, *deployment.Service, deployment.Reader, *pgxpool.Pool) { t.Helper() pool := pgtest.OpenIsolated(t, func(cfg *pgxpool.Config) { dial := cfg.ConnConfig.DialFunc @@ -71,8 +71,8 @@ func delayedReadWriter(t *testing.T, armed *atomic.Bool, reading chan struct{}, t.Error(err) } }) - deployments, operations := testDeployment(t, pool, nil, lease) - return Owner{Lease: lease, Store: store.NewExecution(store.New(pool), lease), Deployment: operations}, deployments, pool + deployments, reader, operations := testDeployment(t, pool, nil, lease) + return Owner{Lease: lease, Store: store.NewExecution(store.New(pool), lease), Deployment: operations}, deployments, reader, pool } func TestSandboxDeploymentDrainPreservesLeaseInFlightRead(t *testing.T) { @@ -89,12 +89,12 @@ func testLifecycleCancellationPreservesLease(t *testing.T, mode string) { var releaseOnce sync.Once unblock := func() { releaseOnce.Do(func() { close(release) }) } defer unblock() - owner, deployments, _ := delayedReadWriter(t, &armed, reading, release) + owner, deployments, reader, _ := delayedReadWriter(t, &armed, reading, release) hub := node.NewHub(node.HubOptions{}) defer hub.Close() id := uuid.NewString() configuration := &RuntimeProvider{InstallationID: id, ProviderKind: "docker", Mode: "nodes", Generation: 1, CoreURL: "https://core.example/api/v1", BackendFingerprint: strings.Repeat("a", 64), Provider: hub.Proxy(uuid.NewString(), "docker", docker.Operations(), 1)} - m, err := newRuntimeManager(owner, deployments, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return configuration, nil })) + m, err := newRuntimeManager(owner, deployments, reader, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return configuration, nil })) if err != nil { t.Fatal(err) } @@ -227,12 +227,12 @@ func (c *delayedCancellationContext) unblock() { } func TestSandboxDeploymentDrainFailureCannotReactivate(t *testing.T) { - _, owner, deployments := resetManagerStore(t) + _, owner, deployments, reader := resetManagerStore(t) hub := node.NewHub(node.HubOptions{}) defer hub.Close() id := uuid.NewString() configuration := &RuntimeProvider{InstallationID: id, ProviderKind: "docker", Mode: "nodes", Generation: 1, CoreURL: "https://core.example/api/v1", BackendFingerprint: strings.Repeat("a", 64), Provider: hub.Proxy(uuid.NewString(), "docker", docker.Operations(), 1)} - m, err := newRuntimeManager(owner, deployments, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return configuration, nil })) + m, err := newRuntimeManager(owner, deployments, reader, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return configuration, nil })) if err != nil { t.Fatal(err) } diff --git a/services/core/internal/execution/sandbox_deployment_setup_test.go b/services/core/internal/execution/sandbox_deployment_setup_test.go index a5b5f59c6..99045edab 100644 --- a/services/core/internal/execution/sandbox_deployment_setup_test.go +++ b/services/core/internal/execution/sandbox_deployment_setup_test.go @@ -25,7 +25,7 @@ func TestDeferredSandboxDeploymentLoadsOnceBeforeNodeCreation(t *testing.T) { var selected atomic.Bool var loads atomic.Int32 configuration := &RuntimeProvider{InstallationID: id, ProviderKind: "docker", CoreURL: "https://core.example/api/v1", BackendFingerprint: strings.Repeat("a", 64), Provider: hub.Proxy(uuid.NewString(), "docker", docker.Operations(), 1)} - m, err := newRuntimeManager(Owner{Lease: heldLease{}}, nil, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { + m, err := newRuntimeManager(Owner{Lease: heldLease{}}, nil, nil, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { loads.Add(1) if !selected.Load() { return nil, nil @@ -70,7 +70,7 @@ func TestDeferredSandboxDeploymentLoadsOnceBeforeNodeCreation(t *testing.T) { func TestDeferredSandboxDeploymentShutdownCancelsLoad(t *testing.T) { entered := make(chan struct{}) - m, err := newRuntimeManager(Owner{Lease: heldLease{}}, nil, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(uuid.NewString(), func(ctx context.Context) (*RuntimeProvider, error) { + m, err := newRuntimeManager(Owner{Lease: heldLease{}}, nil, nil, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(uuid.NewString(), func(ctx context.Context) (*RuntimeProvider, error) { close(entered) <-ctx.Done() return nil, ctx.Err() @@ -95,7 +95,7 @@ func TestDeferredSandboxProviderFailureKeepsRecoveryAvailable(t *testing.T) { available := false loadErr := ErrExecutionUnavailable configuration := &RuntimeProvider{InstallationID: id, ProviderKind: "docker", CoreURL: "https://core.example/api/v1", BackendFingerprint: strings.Repeat("a", 64), Provider: hub.Proxy(uuid.NewString(), "docker", docker.Operations(), 1)} - m, err := newRuntimeManager(Owner{Lease: heldLease{}}, nil, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { + m, err := newRuntimeManager(Owner{Lease: heldLease{}}, nil, nil, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { if !available { return nil, loadErr } @@ -127,7 +127,7 @@ func TestRejectedSandboxCandidatePreservesActiveGeneration(t *testing.T) { id := uuid.NewString() config := &RuntimeProvider{InstallationID: id, ProviderKind: "docker", Mode: "nodes", Generation: 1, CoreURL: "https://core.example/api/v1", BackendFingerprint: strings.Repeat("a", 64), Provider: hub.Proxy(uuid.NewString(), "docker", docker.Operations(), 1)} rejected := errors.New("candidate provider unavailable") - m, err := newRuntimeManager(Owner{Lease: heldLease{}}, unitDeploymentService(t), runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return config, nil }, + m, err := newRuntimeManager(Owner{Lease: heldLease{}}, unitDeploymentService(t), nil, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return config, nil }, func(context.Context, deployment.Setup) (PreparedRuntimeDeployment, error) { return PreparedRuntimeDeployment{}, rejected })) @@ -159,7 +159,7 @@ func TestRejectedSandboxCandidatePreservesActiveGeneration(t *testing.T) { func TestSandboxCandidateValidationDoesNotHoldManagerLock(t *testing.T) { id := uuid.NewString() entered, release := make(chan struct{}), make(chan struct{}) - m, err := newRuntimeManager(Owner{Lease: heldLease{}}, unitDeploymentService(t), runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return nil, nil }, + m, err := newRuntimeManager(Owner{Lease: heldLease{}}, unitDeploymentService(t), nil, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return nil, nil }, func(context.Context, deployment.Setup) (PreparedRuntimeDeployment, error) { close(entered) <-release @@ -192,7 +192,7 @@ func TestCommittedSandboxCandidatePublishesAfterShutdown(t *testing.T) { hub := node.NewHub(node.HubOptions{}) defer hub.Close() id := uuid.NewString() - m, err := newRuntimeManager(Owner{Lease: heldLease{}}, nil, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return nil, nil })) + m, err := newRuntimeManager(Owner{Lease: heldLease{}}, nil, nil, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return nil, nil })) if err != nil { t.Fatal(err) } diff --git a/services/core/internal/execution/sandbox_deployment_switch_test.go b/services/core/internal/execution/sandbox_deployment_switch_test.go index 6a26e6f94..38d6375d9 100644 --- a/services/core/internal/execution/sandbox_deployment_switch_test.go +++ b/services/core/internal/execution/sandbox_deployment_switch_test.go @@ -19,7 +19,7 @@ func TestSandboxManagerSwitchDrainsBeforeDirectActivation(t *testing.T) { defer hub.Close() id := uuid.NewString() config := &RuntimeProvider{InstallationID: id, ProviderKind: "docker", Mode: "nodes", Generation: 1, CoreURL: "https://core.example/api/v1", BackendFingerprint: strings.Repeat("a", 64), Provider: hub.Proxy(uuid.NewString(), "docker", docker.Operations(), 1)} - m, err := newRuntimeManager(Owner{Lease: heldLease{}}, nil, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return config, nil })) + m, err := newRuntimeManager(Owner{Lease: heldLease{}}, nil, nil, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return config, nil })) if err != nil { t.Fatal(err) } @@ -80,7 +80,7 @@ func TestSandboxManagerSwitchDrainsBeforeDirectActivation(t *testing.T) { func TestSandboxManagerFailedActivationStaysPaused(t *testing.T) { id := uuid.NewString() - m, err := newRuntimeManager(Owner{Lease: heldLease{}}, nil, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return nil, errors.New("provider unavailable") })) + m, err := newRuntimeManager(Owner{Lease: heldLease{}}, nil, nil, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return nil, errors.New("provider unavailable") })) if err != nil { t.Fatal(err) } @@ -101,7 +101,7 @@ func TestSandboxManagerCancelledSwitchCannotResumeBeforeDrain(t *testing.T) { defer hub.Close() id := uuid.NewString() config := &RuntimeProvider{InstallationID: id, ProviderKind: "docker", Mode: "nodes", Generation: 1, CoreURL: "https://core.example/api/v1", BackendFingerprint: strings.Repeat("a", 64), Provider: hub.Proxy(uuid.NewString(), "docker", docker.Operations(), 1)} - m, err := newRuntimeManager(Owner{Lease: heldLease{}}, nil, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return config, nil })) + m, err := newRuntimeManager(Owner{Lease: heldLease{}}, nil, nil, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return config, nil })) if err != nil { t.Fatal(err) } @@ -155,7 +155,7 @@ func TestSandboxActivationCannotBypassOutstandingDrain(t *testing.T) { hub := node.NewHub(node.HubOptions{}) defer hub.Close() id := uuid.NewString() - m, err := newRuntimeManager(Owner{Lease: heldLease{}}, nil, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, + m, err := newRuntimeManager(Owner{Lease: heldLease{}}, nil, nil, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return nil, nil }, func(_ context.Context, setup deployment.Setup) (PreparedRuntimeDeployment, error) { return PreparedRuntimeDeployment{Config: &RuntimeProvider{InstallationID: setup.InstallationID, ProviderKind: setup.Provider, Mode: setup.Mode, CoreURL: "https://core.example/api/v1", BackendFingerprint: setup.BackendFingerprint, Provider: hub.Proxy(uuid.NewString(), "docker", docker.Operations(), 1)}}, nil diff --git a/services/core/internal/execution/sandbox_generations_test.go b/services/core/internal/execution/sandbox_generations_test.go index 01c99f724..5a179a8d6 100644 --- a/services/core/internal/execution/sandbox_generations_test.go +++ b/services/core/internal/execution/sandbox_generations_test.go @@ -16,7 +16,7 @@ import ( ) func TestE2BReplacementVerifiesTwiceAndNeverPublishesFailedCommit(t *testing.T) { - _, owner, deployments := resetManagerStore(t) + _, owner, deployments, reader := resetManagerStore(t) id := uuid.NewString() if err := owner.Deployment.Claim(t.Context(), id); err != nil { t.Fatal(err) @@ -50,7 +50,7 @@ func TestE2BReplacementVerifiesTwiceAndNeverPublishesFailedCommit(t *testing.T) }, FenceCredential: func(context.Context) (func(), error) { fenced++; return func() { released++ }, nil }, Publish: func(*RuntimeProvider) { published++ }}, nil }) - m, err := newRuntimeManager(owner, deployments, runtimegateway.NewRegistry(), config) + m, err := newRuntimeManager(owner, deployments, reader, runtimegateway.NewRegistry(), config) if err != nil { t.Fatal(err) } diff --git a/services/core/internal/execution/sandbox_provider_contract_test.go b/services/core/internal/execution/sandbox_provider_contract_test.go index dce51eba7..0c154fe74 100644 --- a/services/core/internal/execution/sandbox_provider_contract_test.go +++ b/services/core/internal/execution/sandbox_provider_contract_test.go @@ -18,7 +18,7 @@ func TestSandboxProviderRegistrationDoesNotRequireAnExecutionVendorBranch(t *tes id := uuid.NewString() config := &RuntimeProvider{InstallationID: id, ProviderKind: "contract-fixture", Mode: mode, CoreURL: "https://core.example/api/v1", BackendFingerprint: strings.Repeat("a", 64), Provider: &lifecycleOnlySandbox{}} - m, err := newRuntimeManager(Owner{Lease: heldLease{}}, nil, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return config, nil })) + m, err := newRuntimeManager(Owner{Lease: heldLease{}}, nil, nil, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return config, nil })) if err != nil { t.Fatal(err) } diff --git a/services/core/internal/execution/sandbox_reset.go b/services/core/internal/execution/sandbox_reset.go index bf5b437b9..187138fd4 100644 --- a/services/core/internal/execution/sandbox_reset.go +++ b/services/core/internal/execution/sandbox_reset.go @@ -11,14 +11,14 @@ import ( "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/store" ) -func (w *Worker) StartSandboxReset(ctx context.Context, input store.SandboxResetRequest) (deployment.View, error) { +func (w *Worker) StartSandboxReset(ctx context.Context, input deployment.ResetRequest) (deployment.View, error) { unlock, err := w.runtimes.lockMutation(ctx) if err != nil { return deployment.View{}, err } defer unlock() m := w.runtimes - if err := m.store.StartSandboxReset(ctx, m.setupInstallationID, input); err != nil { + if err := m.deployment.StartReset(ctx, m.setupInstallationID, input); err != nil { return deployment.View{}, err } return m.committedView(ctx) @@ -31,7 +31,7 @@ func (w *Worker) CancelSandboxReset(ctx context.Context, generation uint64) (dep } defer unlock() m := w.runtimes - if err := m.store.CancelSandboxReset(ctx, m.setupInstallationID, generation); err != nil { + if err := m.deployment.CancelReset(ctx, m.setupInstallationID, generation); err != nil { return deployment.View{}, err } return m.committedView(ctx) @@ -82,7 +82,7 @@ func (m *runtimeManager) resetPage(parent, ctx context.Context) error { return err } defer unlock() - if err := m.store.AdvanceSandboxResetDeadline(ctx); err != nil { + if err := m.deployment.AdvanceResetDeadline(ctx); err != nil { return err } current, err := m.deploymentService.View(ctx) @@ -98,7 +98,7 @@ func (m *runtimeManager) resetPage(parent, ctx context.Context) error { m.resetCursor = "" m.resetRequestedAt = current.Reset.RequestedAt } - candidates, err := m.store.ListSandboxResetSessions(ctx, m.resetCursor, current.Reset.Clear == "force") + candidates, err := m.deploymentReader.ResetSessions(ctx, m.resetCursor, current.Reset.Clear == deployment.ResetForce) if err != nil { return err } @@ -137,7 +137,7 @@ func (m *runtimeManager) resetPage(parent, ctx context.Context) error { } return err } - committed, err := m.store.CompleteSandboxReset(ctx, m.setupInstallationID, current.Generation, current.Reset.RequestedAt) + committed, err := m.deployment.CompleteReset(ctx, m.setupInstallationID, current.Generation, current.Reset.RequestedAt) if err != nil { recovery := m.restoreCommittedDeployment() if recovery != nil { diff --git a/services/core/internal/execution/sandbox_reset_test.go b/services/core/internal/execution/sandbox_reset_test.go index d4f3b8ff6..0c90d6093 100644 --- a/services/core/internal/execution/sandbox_reset_test.go +++ b/services/core/internal/execution/sandbox_reset_test.go @@ -26,22 +26,22 @@ import ( // The execution lease is database-scoped, so these manager tests own a database. // They receive the pooled Store, the Owner of its execution lease, which the -// test closes when it ends, and the pooled deployment service the Worker -// receives beside it. -func resetManagerStore(t *testing.T) (*store.Store, Owner, *deployment.Service) { +// test closes when it ends, and the pooled deployment service and reader the +// Worker receives beside it. +func resetManagerStore(t *testing.T) (*store.Store, Owner, *deployment.Service, deployment.Reader) { t.Helper() return resetManagerStoreConfig(t, nil) } -func resetManagerStoreConfig(t *testing.T, configure func(*pgxpool.Config)) (*store.Store, Owner, *deployment.Service) { +func resetManagerStoreConfig(t *testing.T, configure func(*pgxpool.Config)) (*store.Store, Owner, *deployment.Service, deployment.Reader) { t.Helper() - s, owner, deployments, _ := resetManagerStoreDB(t, configure) - return s, owner, deployments + s, owner, deployments, reader, _ := resetManagerStoreDB(t, configure) + return s, owner, deployments, reader } // resetManagerStoreDB also returns the test database, for tests that build // adapters on it. -func resetManagerStoreDB(t *testing.T, configure func(*pgxpool.Config)) (*store.Store, Owner, *deployment.Service, *pgxpool.Pool) { +func resetManagerStoreDB(t *testing.T, configure func(*pgxpool.Config)) (*store.Store, Owner, *deployment.Service, deployment.Reader, *pgxpool.Pool) { t.Helper() pool := pgtest.OpenIsolated(t, configure) cipher, err := credentialcrypto.New(bytes.Repeat([]byte{8}, 32)) @@ -49,27 +49,27 @@ func resetManagerStoreDB(t *testing.T, configure func(*pgxpool.Config)) (*store. t.Fatal(err) } s := store.NewWithCredentialCipher(pool, cipher) - owner, deployments := testOwner(t, pool, cipher, s) - return s, owner, deployments, pool + owner, deployments, reader := testOwner(t, pool, cipher, s) + return s, owner, deployments, reader, pool } // testOwner acquires the execution lease on pool and builds s's execution // writer and the deployment execution operations on it, and the pooled -// deployment service, as cmd/server does. The lease closes when the test ends. -func testOwner(t *testing.T, pool *pgxpool.Pool, cipher *credentialcrypto.Cipher, s *store.Store) (Owner, *deployment.Service) { +// deployment service and reader, as cmd/server does. The lease closes when the +// test ends. +func testOwner(t *testing.T, pool *pgxpool.Pool, cipher *credentialcrypto.Cipher, s *store.Store) (Owner, *deployment.Service, deployment.Reader) { t.Helper() lease, err := pgunit.AcquireLease(t.Context(), pool) if err != nil { t.Fatal(err) } t.Cleanup(func() { _ = lease.Close(context.Background()) }) - deployments, operations := testDeployment(t, pool, cipher, lease) - return Owner{Lease: lease, Store: store.NewExecution(s, lease), Deployment: operations}, deployments + deployments, reader, operations := testDeployment(t, pool, cipher, lease) + return Owner{Lease: lease, Store: store.NewExecution(s, lease), Deployment: operations}, deployments, reader } func TestSandboxResetPageTimeoutRecoversCommittedOwner(t *testing.T) { - _, owner, deployments := resetManagerStore(t) - w := owner.Store + _, owner, deployments, reader := resetManagerStore(t) id := initializeE2BDeployment(t, owner) hub := node.NewHub(node.HubOptions{}) defer hub.Close() @@ -85,7 +85,7 @@ func TestSandboxResetPageTimeoutRecoversCommittedOwner(t *testing.T) { } return &RuntimeProvider{InstallationID: id, ProviderKind: setup.Provider, Mode: setup.Mode, Generation: setup.Generation, CoreURL: "https://core.example/api/v1", BackendFingerprint: setup.BackendFingerprint, Provider: hub.Proxy(uuid.NewString(), "docker", docker.Operations(), 1)}, nil }) - m, err := newRuntimeManager(owner, deployments, runtimegateway.NewRegistry(), config) + m, err := newRuntimeManager(owner, deployments, reader, runtimegateway.NewRegistry(), config) if err != nil { t.Fatal(err) } @@ -108,7 +108,7 @@ func TestSandboxResetPageTimeoutRecoversCommittedOwner(t *testing.T) { } }() audit := adminaudit.WithSource(t.Context(), adminaudit.Source{CredentialID: "reset-test", ActorLabel: "operator", RequestID: "request", TraceID: "trace"}) - if err := w.StartSandboxReset(audit, id, store.SandboxResetRequest{ExpectedGeneration: 1, Clear: "force"}); err != nil { + if err := owner.Deployment.StartReset(audit, id, deployment.ResetRequest{ExpectedGeneration: 1, Clear: deployment.ResetForce}); err != nil { t.Fatal(err) } reset, err := deployments.View(t.Context()) @@ -174,7 +174,7 @@ func initializeE2BDeployment(t *testing.T, owner Owner) string { } func TestSandboxResetPublishesCommittedGenerationWithoutReading(t *testing.T) { - _, owner, pooled, pool := resetManagerStoreDB(t, nil) + _, owner, pooled, _, pool := resetManagerStoreDB(t, nil) id := initializeE2BDeployment(t, owner) // The manager reads the deployment through a reader that fails every read // of the committed reset, so publication cannot depend on one. @@ -189,7 +189,7 @@ func TestSandboxResetPublishesCommittedGenerationWithoutReading(t *testing.T) { } return snapshot, err }} - deployments, operations := deploymentOperations(t, &strictDeploymentStorage{t: t}, reader, &strictExecutionStorage{t: t}) + deployments, _ := deploymentOperations(t, &strictDeploymentStorage{t: t}, reader, &strictExecutionStorage{t: t}) hub := node.NewHub(node.HubOptions{}) defer hub.Close() config := NewDeferredRuntimeProvider(id, func(ctx context.Context) (*RuntimeProvider, error) { @@ -206,7 +206,7 @@ func TestSandboxResetPublishesCommittedGenerationWithoutReading(t *testing.T) { } published = append(published, generation) } - m, err := newRuntimeManager(Owner{Lease: owner.Lease, Store: owner.Store, Deployment: operations}, deployments, runtimegateway.NewRegistry(), config) + m, err := newRuntimeManager(owner, deployments, adapter, runtimegateway.NewRegistry(), config) if err != nil { t.Fatal(err) } @@ -215,7 +215,7 @@ func TestSandboxResetPublishesCommittedGenerationWithoutReading(t *testing.T) { t.Fatal(err) } audit := adminaudit.WithSource(t.Context(), adminaudit.Source{CredentialID: "reset-test", ActorLabel: "operator", RequestID: "request", TraceID: "trace"}) - if err := owner.Store.StartSandboxReset(audit, id, store.SandboxResetRequest{ExpectedGeneration: 1, Clear: "force"}); err != nil { + if err := owner.Deployment.StartReset(audit, id, deployment.ResetRequest{ExpectedGeneration: 1, Clear: deployment.ResetForce}); err != nil { t.Fatal(err) } if err := m.resetStep(t.Context()); err != nil { @@ -241,7 +241,7 @@ func TestCommittedResetViewStopsOwnerWithoutLease(t *testing.T) { return deployment.Snapshot{Record: deployment.Record{InstallationID: id, WebManaged: true, Generation: 1}}, nil }} deployments, operations := deploymentOperations(t, &strictDeploymentStorage{t: t}, reader, &strictExecutionStorage{t: t}) - m, err := newRuntimeManager(Owner{Lease: lostLease{}, Deployment: operations}, deployments, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return nil, nil })) + m, err := newRuntimeManager(Owner{Lease: lostLease{}, Deployment: operations}, deployments, reader, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return nil, nil })) if err != nil { t.Fatal(err) } @@ -261,16 +261,16 @@ func TestCommittedResetViewStopsOwnerWithoutLease(t *testing.T) { } func TestSandboxResetChangesReturnViewReadAfterCommit(t *testing.T) { - _, owner, deployments := resetManagerStore(t) + _, owner, deployments, reader := resetManagerStore(t) id := initializeE2BDeployment(t, owner) - m, err := newRuntimeManager(owner, deployments, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return nil, nil })) + m, err := newRuntimeManager(owner, deployments, reader, runtimegateway.NewRegistry(), NewDeferredRuntimeProvider(id, func(context.Context) (*RuntimeProvider, error) { return nil, nil })) if err != nil { t.Fatal(err) } defer func() { m.stop(); m.drain() }() worker := &Worker{runtimes: m} audit := adminaudit.WithSource(t.Context(), adminaudit.Source{CredentialID: "reset-test", ActorLabel: "operator", RequestID: "request", TraceID: "trace"}) - started, err := worker.StartSandboxReset(audit, store.SandboxResetRequest{ExpectedGeneration: 1, Clear: "force"}) + started, err := worker.StartSandboxReset(audit, deployment.ResetRequest{ExpectedGeneration: 1, Clear: deployment.ResetForce}) if err != nil { t.Fatal(err) } diff --git a/services/core/internal/execution/sandbox_snapshot_budget_test.go b/services/core/internal/execution/sandbox_snapshot_budget_test.go index a17955865..9411356cb 100644 --- a/services/core/internal/execution/sandbox_snapshot_budget_test.go +++ b/services/core/internal/execution/sandbox_snapshot_budget_test.go @@ -9,10 +9,10 @@ import ( "time" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/adminaudit" + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/deployment" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/runtimegateway" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sandbox/docker" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sandbox/node" - "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/store" "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" @@ -60,11 +60,10 @@ func (d *snapshotBudget) TraceQueryEnd(ctx context.Context, _ *pgx.Conn, data pg func TestSandboxResetSnapshotFitsPageBudget(t *testing.T) { budget := &snapshotBudget{t: t} - _, owner, deployments := resetManagerStoreConfig(t, func(cfg *pgxpool.Config) { + _, owner, deployments, reader := resetManagerStoreConfig(t, func(cfg *pgxpool.Config) { cfg.ConnConfig.RuntimeParams["jit"] = "on" cfg.ConnConfig.Tracer = budget }) - w := owner.Store id := initializeE2BDeployment(t, owner) hub := node.NewHub(node.HubOptions{}) defer hub.Close() @@ -75,7 +74,7 @@ func TestSandboxResetSnapshotFitsPageBudget(t *testing.T) { } return &RuntimeProvider{InstallationID: id, ProviderKind: setup.Provider, Generation: setup.Generation, Mode: setup.Mode, CoreURL: "https://core.example/api/v1", BackendFingerprint: setup.BackendFingerprint, Provider: hub.Proxy(uuid.NewString(), "docker", docker.Operations(), 1)}, nil }) - m, err := newRuntimeManager(owner, deployments, runtimegateway.NewRegistry(), configuration) + m, err := newRuntimeManager(owner, deployments, reader, runtimegateway.NewRegistry(), configuration) if err != nil { t.Fatal(err) } @@ -83,7 +82,7 @@ func TestSandboxResetSnapshotFitsPageBudget(t *testing.T) { if _, err = m.ensureDeployment(t.Context()); err != nil { t.Fatal(err) } - if err = w.StartSandboxReset(adminaudit.WithSource(t.Context(), adminaudit.Source{CredentialID: "reset-test", ActorLabel: "operator", RequestID: "request", TraceID: "trace"}), id, store.SandboxResetRequest{ExpectedGeneration: 1, Clear: "force"}); err != nil { + if err = owner.Deployment.StartReset(adminaudit.WithSource(t.Context(), adminaudit.Source{CredentialID: "reset-test", ActorLabel: "operator", RequestID: "request", TraceID: "trace"}), id, deployment.ResetRequest{ExpectedGeneration: 1, Clear: deployment.ResetForce}); err != nil { t.Fatal(err) } budget.armed.Store(true) diff --git a/services/core/internal/execution/worker.go b/services/core/internal/execution/worker.go index 668eb6a27..b7ccc1f88 100644 --- a/services/core/internal/execution/worker.go +++ b/services/core/internal/execution/worker.go @@ -59,6 +59,9 @@ func StartWorker(ctx context.Context, dispatcher *Dispatcher, owner Owner) (_ *W if dispatcher.Deployment == nil { return nil, errors.New("execution worker requires the deployment service") } + if dispatcher.DeploymentReader == nil { + return nil, errors.New("execution worker requires the deployment reader") + } if dispatcher.Sessions == nil { return nil, errors.New("execution worker requires the Session service") } @@ -75,7 +78,7 @@ func StartWorker(ctx context.Context, dispatcher *Dispatcher, owner Owner) (_ *W owned.Store = owner.Store owned.notifications = &executionNotifications{} worker := &Worker{concurrency: dispatcher.MaxConcurrentExecutions, dispatcher: &owned, admission: dispatcher.Store, lease: owner.Lease, directoryReads: make(chan directoryReadRequest), fileWrites: make(chan fileWriteRequest), stopped: make(chan struct{}), scheduleWake: make(chan struct{}, 1), enrolledConnections: make(map[string]*runtimeConnection)} - worker.runtimes, err = newRuntimeManager(owner, owned.Deployment, owned.Registry, owned.ManagedRuntimes) + worker.runtimes, err = newRuntimeManager(owner, owned.Deployment, owned.DeploymentReader, owned.Registry, owned.ManagedRuntimes) if err != nil { return nil, err } diff --git a/services/core/internal/persistence/postgres/deploymentpg/reset_test.go b/services/core/internal/persistence/postgres/deploymentpg/reset_test.go new file mode 100644 index 000000000..9c46bb524 --- /dev/null +++ b/services/core/internal/persistence/postgres/deploymentpg/reset_test.go @@ -0,0 +1,172 @@ +package deploymentpg_test + +import ( + "errors" + "slices" + "testing" + "time" + + "github.com/google/uuid" + + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/adminaudit" + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/deployment" + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/persistence/postgres/pgunit" + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sandbox" +) + +// resetAudit lists the audited deployment actions in order, each with the +// administrator credential that recorded it. +func resetAudit(t *testing.T, f fixture) []string { + t.Helper() + rows, err := f.pool.Query(t.Context(), "SELECT action || ' ' || admin_credential_id FROM admin_audit_log WHERE resource_type='sandbox_deployment' ORDER BY created_at, id") + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var entries []string + for rows.Next() { + var entry string + if err := rows.Scan(&entry); err != nil { + t.Fatal(err) + } + entries = append(entries, entry) + } + if err := rows.Err(); err != nil { + t.Fatal(err) + } + return entries +} + +func resetCount(t *testing.T, f fixture, query string) int { + t.Helper() + var count int + if err := f.pool.QueryRow(t.Context(), query).Scan(&count); err != nil { + t.Fatal(err) + } + return count +} + +// A reset keeps its original request time and deadline across a replay and an +// escalation, and completion clears the deployment under the starter's audit +// source. +func TestResetStartEscalateAndComplete(t *testing.T) { + f := newFixture(t) + changes, _ := f.execution(t) + installation, view := f.initialize(t, changes, sandbox.Selection{Provider: "docker", DeploymentSpec: testSpecification("docker")}) + f.enroll(t, view, deployment.Capacity{MaxActive: 1, MaxRetained: 1}) + if _, err := f.service.CreateEnrollment(admin(t), deployment.Capacity{MaxActive: 1, MaxRetained: 1}); err != nil { + t.Fatal(err) + } + + auto := deployment.ResetRequest{ExpectedGeneration: view.Generation, Clear: deployment.ResetAuto} + if err := changes.StartReset(admin(t), installation, auto); err != nil { + t.Fatal(err) + } + started, err := f.service.View(t.Context()) + if err != nil || started.Reset == nil || started.Reset.DeadlineAt == nil || started.Reset.DeadlineAt.Sub(started.Reset.RequestedAt) != time.Hour || started.Reset.ForcedAt != nil { + t.Fatalf("started reset = %+v, %v", started.Reset, err) + } + if err := changes.StartReset(admin(t), installation, auto); err != nil { + t.Fatal(err) + } + force := auto + force.Clear = deployment.ResetForce + if err := changes.StartReset(adminaudit.WithSource(t.Context(), adminaudit.Source{CredentialID: "escalating-admin", RequestID: uuid.NewString(), TraceID: uuid.NewString()}), installation, force); err != nil { + t.Fatal(err) + } + forced, err := f.service.View(t.Context()) + if err != nil || forced.Reset == nil || forced.Reset.Clear != deployment.ResetForce || forced.Reset.ForcedAt == nil || + !forced.Reset.RequestedAt.Equal(started.Reset.RequestedAt) || !forced.Reset.DeadlineAt.Equal(*started.Reset.DeadlineAt) { + t.Fatalf("forced reset = %+v, %v", forced.Reset, err) + } + + // Completion carries no caller source: it audits as the administrator who + // started the reset. + committed, err := changes.CompleteReset(t.Context(), installation, view.Generation, forced.Reset.RequestedAt) + if err != nil || committed != view.Generation+1 { + t.Fatalf("CompleteReset = %d, %v", committed, err) + } + cleared, err := f.service.View(t.Context()) + if err != nil || cleared.Generation != committed || cleared.Provider != "" || cleared.Reset != nil || cleared.OwnerEpoch != view.OwnerEpoch+1 { + t.Fatalf("cleared deployment = %+v, %v", cleared, err) + } + if nodes := resetCount(t, f, "SELECT count(*) FROM runtime_nodes WHERE removed_at IS NULL"); nodes != 0 { + t.Fatalf("%d nodes survived the reset", nodes) + } + if tokens := resetCount(t, f, "SELECT count(*) FROM runtime_node_enrollments WHERE consumed_at IS NULL AND expires_at > clock_timestamp()"); tokens != 0 { + t.Fatalf("%d enrollment tokens survived the reset", tokens) + } + if got, want := resetAudit(t, f), []string{"reset_start fixture-admin", "reset_force escalating-admin", "reset_complete fixture-admin"}; !slices.Equal(got, want) { + t.Fatalf("audit = %q, want %q", got, want) + } +} + +// A passed auto deadline escalates under the starter's source, and a +// cancellation restores admission. +func TestResetDeadlineAndCancel(t *testing.T) { + f := newFixture(t) + changes, _ := f.execution(t) + installation, view := f.initialize(t, changes, sandbox.Selection{Provider: "docker", DeploymentSpec: testSpecification("docker")}) + if err := changes.StartReset(admin(t), installation, deployment.ResetRequest{ExpectedGeneration: view.Generation, Clear: deployment.ResetAuto}); err != nil { + t.Fatal(err) + } + if err := changes.AdvanceResetDeadline(t.Context()); err != nil { + t.Fatal(err) + } + if _, err := f.pool.Exec(t.Context(), "UPDATE runtime_deployment SET reset_deadline_at = clock_timestamp() - interval '1 second'"); err != nil { + t.Fatal(err) + } + if err := changes.AdvanceResetDeadline(t.Context()); err != nil { + t.Fatal(err) + } + if forced, err := f.service.View(t.Context()); err != nil || forced.Reset == nil || forced.Reset.Clear != deployment.ResetForce { + t.Fatalf("reset after its deadline = %+v, %v", forced.Reset, err) + } + if err := changes.CancelReset(admin(t), installation, view.Generation); err != nil { + t.Fatal(err) + } + if paused := resetCount(t, f, "SELECT count(*) FROM runtime_deployment WHERE admission_paused OR reset_clear IS NOT NULL OR reset_audit IS NOT NULL"); paused != 0 { + t.Fatal("cancellation left the reset or paused admission") + } + if got, want := resetAudit(t, f), []string{"reset_start fixture-admin", "reset_deadline fixture-admin", "reset_cancel fixture-admin"}; !slices.Equal(got, want) { + t.Fatalf("audit = %q, want %q", got, want) + } +} + +// A rejected audit entry and a lost lease leave no reset behind. +func TestResetWritesNothingWithoutAuditOrLease(t *testing.T) { + f := newFixture(t) + closed := setupClosedExecution(t, f) + changes, _ := f.execution(t) + installation, view := f.initialize(t, changes, sandbox.Selection{Provider: "docker", DeploymentSpec: testSpecification("docker")}) + request := deployment.ResetRequest{ExpectedGeneration: view.Generation, Clear: deployment.ResetAuto} + if err := changes.StartReset(adminaudit.WithSource(t.Context(), adminaudit.Source{}), installation, request); !errors.Is(err, adminaudit.ErrInvalidSource) { + t.Fatalf("StartReset with an invalid source = %v", err) + } + if err := closed.StartReset(admin(t), installation, request); !errors.Is(err, pgunit.ErrLeaseClosed) { + t.Fatalf("StartReset on a closed lease = %v", err) + } + if paused := resetCount(t, f, "SELECT count(*) FROM runtime_deployment WHERE admission_paused OR reset_clear IS NOT NULL"); paused != 0 { + t.Fatal("a rejected reset paused admission") + } +} + +func TestResetSessionsAndAddressBindings(t *testing.T) { + f := newFixture(t) + if _, err := f.adapter.ResetSessions(t.Context(), "not-a-session", false); !errors.Is(err, deployment.ErrInvalidInput) { + t.Fatalf("ResetSessions with a malformed cursor = %v", err) + } + if sessions, err := f.adapter.ResetSessions(t.Context(), uuid.NewString(), true); err != nil || len(sessions) != 0 { + t.Fatalf("ResetSessions without Sessions = %v, %v", sessions, err) + } + + changes, _ := f.execution(t) + _, view := f.initialize(t, changes, sandbox.Selection{Provider: "docker", DeploymentSpec: testSpecification("docker")}) + f.enroll(t, view, deployment.Capacity{MaxActive: 1, MaxRetained: 1}) + if bindings, err := f.service.AddressBindings(t.Context()); err != nil || bindings != (deployment.AddressBindings{Nodes: 1}) { + t.Fatalf("AddressBindings = %+v, %v", bindings, err) + } + if bindings, err := f.adapter.AddressBindings(t.Context(), "https://other.example"); err != nil || bindings != (deployment.AddressBindings{Nodes: 1, NodesOnOtherAddress: 1}) { + t.Fatalf("AddressBindings for another address = %+v, %v", bindings, err) + } +} diff --git a/services/core/internal/persistence/postgres/deploymentpg/store.go b/services/core/internal/persistence/postgres/deploymentpg/store.go index 40117d569..decc3aaa9 100644 --- a/services/core/internal/persistence/postgres/deploymentpg/store.go +++ b/services/core/internal/persistence/postgres/deploymentpg/store.go @@ -274,6 +274,47 @@ func (s *Store) DisconnectNode(ctx context.Context, nodeID, connectionID string, }) } +func (s *Store) ResetSessions(ctx context.Context, after string, force bool) ([]deployment.ResetSession, error) { + cursor := pgtype.UUID{Valid: true} + if after != "" { + var err error + if cursor, err = parseID(after); err != nil { + return nil, err + } + } + rows, err := s.pool.Queries().ListSandboxResetSessions(ctx, sqlc.ListSandboxResetSessionsParams{AfterID: cursor, Force: force}) + if err != nil { + return nil, err + } + result := make([]deployment.ResetSession, 0, len(rows)) + for _, row := range rows { + result = append(result, deployment.ResetSession{SessionID: uuidString(row.ID), TenantID: uuidString(row.TenantID)}) + } + return result, nil +} + +func (s *Store) AddressBindings(ctx context.Context, publicURL string) (deployment.AddressBindings, error) { + var result deployment.AddressBindings + err := s.pool.Snapshot(ctx, func(ctx context.Context, tx pgx.Tx) error { + q := sqlc.New(tx) + bindings, err := q.CountAddressBindings(ctx, publicURL) + if err != nil { + return err + } + resources, err := q.CountRuntimeDeploymentResources(ctx) + if err != nil { + return err + } + result = deployment.AddressBindings{Nodes: bindings.Nodes, NodesOnOtherAddress: bindings.NodesOnOtherAddress, + HostedSandboxes: resources.Allocations + resources.Pending, SelfHostedExecutors: bindings.SelfHostedExecutors} + return nil + }) + if err != nil { + return deployment.AddressBindings{}, err + } + return result, nil +} + func (s *Store) SampleHostHistory(ctx context.Context) (int64, error) { return s.pool.Queries().SampleNodeHostHistory(ctx) } diff --git a/services/core/internal/persistence/postgres/deploymentpg/tx.go b/services/core/internal/persistence/postgres/deploymentpg/tx.go index 4c78ffa32..fbab74148 100644 --- a/services/core/internal/persistence/postgres/deploymentpg/tx.go +++ b/services/core/internal/persistence/postgres/deploymentpg/tx.go @@ -11,6 +11,7 @@ import ( "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgtype" + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/adminaudit" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/credentialcrypto" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/db/sqlc" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/deployment" @@ -324,6 +325,58 @@ func (t *deploymentTx) RetainGeneration() error { return t.q.RetainSandboxGenera func (t *deploymentTx) CollectGenerations() error { return t.q.CollectSandboxGenerations(t.ctx) } +func (t *deploymentTx) StartReset(clear string, deadlineSeconds int32, source adminaudit.Source) error { + // The audit entry recorded in the same transaction validates the source + // before commit. Only the typed, non-secret source is stored. + audit, err := json.Marshal(source) + if err != nil { + return err + } + return t.q.StartSandboxReset(t.ctx, sqlc.StartSandboxResetParams{Clear: pgtype.Text{String: clear, Valid: true}, DeadlineSeconds: deadlineSeconds, Audit: audit}) +} + +func (t *deploymentTx) ForceReset() error { return t.q.ForceSandboxReset(t.ctx) } + +func (t *deploymentTx) CancelReset() error { return t.q.CancelSandboxReset(t.ctx) } + +func (t *deploymentTx) LoadResetSource() (adminaudit.Source, error) { + d, err := t.loadDeployment() + if err != nil { + return adminaudit.Source{}, err + } + return ResetSource(d) +} + +var errNoResetSource = errors.New("sandbox reset has no stored administrator source") + +// ResetSource decodes the administrator source that started the running reset +// from the stored deployment row. Audit entries the reset records later carry +// that source. +func ResetSource(d sqlc.RuntimeDeployment) (adminaudit.Source, error) { + var source adminaudit.Source + if !d.ResetClear.Valid || json.Unmarshal(d.ResetAudit, &source) != nil { + return adminaudit.Source{}, errNoResetSource + } + return source, nil +} + +func (t *deploymentTx) CompleteReset() error { + if err := t.q.CompleteSandboxReset(t.ctx); err != nil { + return err + } + if err := t.q.RetireSandboxNodes(t.ctx); err != nil { + return err + } + if err := t.q.RetireSandboxEnrollments(t.ctx); err != nil { + return err + } + return t.q.ClearSandboxGenerations(t.ctx) +} + func (t *deploymentTx) RecordAudit(action, installationID string) error { return auditpg.RecordDeploymentMutation(t.ctx, t.q, action, "sandbox_deployment", installationID) } + +func (t *deploymentTx) RecordAuditAs(source adminaudit.Source, action, installationID string) error { + return auditpg.RecordDeploymentMutation(adminaudit.WithSource(t.ctx, source), t.q, action, "sandbox_deployment", installationID) +} diff --git a/services/core/internal/sandbox/providers/configuration_flow_test.go b/services/core/internal/sandbox/providers/configuration_flow_test.go index 144a3647a..a37977ccf 100644 --- a/services/core/internal/sandbox/providers/configuration_flow_test.go +++ b/services/core/internal/sandbox/providers/configuration_flow_test.go @@ -120,7 +120,7 @@ func TestAdditionalConfigurationProviderUsesCommonAPIAndStore(t *testing.T) { // The flow reaches only the store areas and the deployment setup; every // other dependency panics if called. h, err := api.NewHandler(api.Dependencies{ - Engine: "codex", CoreKeys: auth, InstallationBindings: s, + Engine: "codex", CoreKeys: auth, InstallationBindings: service, Projects: struct{ api.Projects }{}, ProjectsReader: struct{ api.ProjectsReader }{}, ModelProviders: struct{ api.ModelProviders }{}, ModelProvidersReader: struct{ api.ModelProvidersReader }{}, Vaults: struct{ api.Vaults }{}, VaultsReader: struct{ api.VaultsReader }{}, @@ -189,7 +189,7 @@ func (l leaseSetup) UpdateSandboxDeployment(context.Context, sandbox.Selection) return deployment.View{}, nil } -func (l leaseSetup) StartSandboxReset(context.Context, store.SandboxResetRequest) (deployment.View, error) { +func (l leaseSetup) StartSandboxReset(context.Context, deployment.ResetRequest) (deployment.View, error) { l.t.Fatal("unexpected call to StartSandboxReset") return deployment.View{}, nil } diff --git a/services/core/internal/store/admin_session_archive.go b/services/core/internal/store/admin_session_archive.go index 73f041409..2ee7157cb 100644 --- a/services/core/internal/store/admin_session_archive.go +++ b/services/core/internal/store/admin_session_archive.go @@ -9,6 +9,7 @@ import ( "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/adminaudit" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/persistence/postgres/auditpg" + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/persistence/postgres/deploymentpg" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/persistence/postgres/pgunit" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/persistence/postgres/sessionpg" @@ -85,7 +86,7 @@ func (s *Store) archiveManagedSession(ctx context.Context, tenantID, sessionID s result, err = getManagedSessionArchive(ctx, q, tenant, session) return err } - source, err := sandboxResetAudit(current) + source, err := deploymentpg.ResetSource(current) if err != nil { return err } diff --git a/services/core/internal/store/fixture_db_test.go b/services/core/internal/store/fixture_db_test.go index 743c9f77a..7ce0713d6 100644 --- a/services/core/internal/store/fixture_db_test.go +++ b/services/core/internal/store/fixture_db_test.go @@ -9,6 +9,7 @@ import ( "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/credentialcrypto" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/execution" + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/persistence/postgres/deploymentpg" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/persistence/postgres/modelconfigurationpg" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/persistence/postgres/pgunit" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/store" @@ -44,8 +45,9 @@ func newManagedTestStoreDB(t *testing.T) (*store.Store, fixtureDB) { // execution lease on db and hands it, with the execution writer built on it, to // the Worker, which closes it when Run exits. The Worker opens MCP bearer tokens // through the vaults service on db, records model configuration observations -// through the model configuration adapter on db and runs Session use cases -// and reads through the Session service and adapter on db. +// through the model configuration adapter on db, runs Session use cases and +// reads through the Session service and adapter on db, and reads the +// deployment through the deployment adapter on db. func startWorker(t testing.TB, ctx context.Context, db fixtureDB, dispatcher *execution.Dispatcher) *execution.Worker { t.Helper() worker, err := startWorkerErr(ctx, db, dispatcher) @@ -79,6 +81,7 @@ func startWorkerErr(ctx context.Context, db fixtureDB, dispatcher *execution.Dis owned.Deployment = deployments owned.Sessions = sessionService owned.SessionsReader = sessionStore + owned.DeploymentReader = deploymentpg.New(pgunit.NewPool(db.pool), db.cipher) return execution.StartWorker(ctx, &owned, execution.Owner{ Lease: lease, Store: store.NewExecution(dispatcher.Store, lease), diff --git a/services/core/internal/store/public_handler_fixture_test.go b/services/core/internal/store/public_handler_fixture_test.go index 6300ed783..04a15361d 100644 --- a/services/core/internal/store/public_handler_fixture_test.go +++ b/services/core/internal/store/public_handler_fixture_test.go @@ -76,7 +76,7 @@ func publicHandler(t testing.TB, s *store.Store, db fixtureDB, keys fixtureKeyRe return nil, err } deps := api.Dependencies{ - Engine: engine, CoreKeys: admin, InstallationBindings: s, + Engine: engine, CoreKeys: admin, InstallationBindings: fixtureDeployment(t, db), Projects: projectService, ProjectsReader: fixtureProjectsReader{Reader: projectStore, keys: keys}, ModelProviders: modelConfigurationService, ModelProvidersReader: modelConfigurationStore, Vaults: vaultService, VaultsReader: vaultStore, @@ -290,7 +290,7 @@ func (s strictStandIn) UpdateSandboxDeployment(context.Context, sandbox.Selectio return deployment.View{}, nil } -func (s strictStandIn) StartSandboxReset(context.Context, store.SandboxResetRequest) (deployment.View, error) { +func (s strictStandIn) StartSandboxReset(context.Context, deployment.ResetRequest) (deployment.View, error) { s.unexpected("StartSandboxReset") return deployment.View{}, nil } diff --git a/services/core/internal/store/public_url.go b/services/core/internal/store/public_url.go index 22e9beedf..d3a6ee70c 100644 --- a/services/core/internal/store/public_url.go +++ b/services/core/internal/store/public_url.go @@ -1,33 +1,5 @@ package store -import ( - "context" -) - // SetPublicURL records OAC_PUBLIC_URL, validated by the caller. Placement // admits only nodes enrolled with it. Call it once, before serving requests. func (s *Store) SetPublicURL(value string) { s.publicURL = value } - -// AddressBindings counts what is bound to an installation address: nodes -// connect to the address they enrolled with, hosted sandboxes were started with -// the address current at the time, and self-hosted executors were installed -// with an advertised remote_url. -type AddressBindings struct { - Nodes int64 `json:"nodes"` - NodesOnOtherAddress int64 `json:"nodes_on_other_address"` - HostedSandboxes int64 `json:"hosted_sandboxes"` - SelfHostedExecutors int64 `json:"self_hosted_executors"` -} - -func (s *Store) AddressBindings(ctx context.Context) (AddressBindings, error) { - bindings, err := s.queries.CountAddressBindings(ctx, s.publicURL) - if err != nil { - return AddressBindings{}, err - } - resources, err := s.queries.CountRuntimeDeploymentResources(ctx) - if err != nil { - return AddressBindings{}, err - } - return AddressBindings{Nodes: bindings.Nodes, NodesOnOtherAddress: bindings.NodesOnOtherAddress, - HostedSandboxes: resources.Allocations + resources.Pending, SelfHostedExecutors: bindings.SelfHostedExecutors}, nil -} diff --git a/services/core/internal/store/runtime_deployment.go b/services/core/internal/store/runtime_deployment.go index 8eae86660..080d33178 100644 --- a/services/core/internal/store/runtime_deployment.go +++ b/services/core/internal/store/runtime_deployment.go @@ -3,6 +3,7 @@ package store import ( "context" "encoding/json" + "errors" "fmt" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/db/sqlc" @@ -11,6 +12,10 @@ import ( "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sessions" ) +// ErrSandboxResetAdmission rejects new hosted work while a sandbox reset +// pauses admission. +var ErrSandboxResetAdmission = errors.New("hosted admission is paused for a sandbox reset") + // sandboxProviders interprets the deployment's provider declarations for // Session admission and placement until they move out of the Store. var sandboxProviders = providers.Builtin() diff --git a/services/core/internal/store/runtime_node_capacity_test.go b/services/core/internal/store/runtime_node_capacity_test.go index 3c8a8c643..c76613475 100644 --- a/services/core/internal/store/runtime_node_capacity_test.go +++ b/services/core/internal/store/runtime_node_capacity_test.go @@ -13,7 +13,7 @@ func TestRuntimeEnrollmentCapacityPrecedesResetConflict(t *testing.T) { nodes := deploymentService(t, s) // Enter reset through its owner transaction so the fixture obeys the // admission invariant. - if err := w.StartSandboxReset(SandboxResetTestContext(t.Context()), view.InstallationID, SandboxResetRequest{ExpectedGeneration: view.Generation, Clear: "auto"}); err != nil { + if err := deploymentExecution(t, w).StartReset(SandboxResetTestContext(t.Context()), view.InstallationID, deployment.ResetRequest{ExpectedGeneration: view.Generation, Clear: "auto"}); err != nil { t.Fatal(err) } if _, err := nodes.CreateEnrollment(t.Context(), deployment.Capacity{MaxActive: 1, MaxRetained: 3}); !errors.Is(err, deployment.ErrResetInProgress) { diff --git a/services/core/internal/store/sandbox_deployment_resources_test.go b/services/core/internal/store/sandbox_deployment_resources_test.go index 34f546bcf..6368615db 100644 --- a/services/core/internal/store/sandbox_deployment_resources_test.go +++ b/services/core/internal/store/sandbox_deployment_resources_test.go @@ -50,9 +50,9 @@ func TestSandboxDeploymentMutationViewsIncludeActualResources(t *testing.T) { for _, maintenance := range []bool{true, false} { var err error if maintenance { - err = w.StartSandboxReset(SandboxResetTestContext(t.Context()), installation, SandboxResetRequest{ExpectedGeneration: 1, Clear: "auto"}) + err = deploymentExecution(t, w).StartReset(SandboxResetTestContext(t.Context()), installation, deployment.ResetRequest{ExpectedGeneration: 1, Clear: "auto"}) } else { - err = w.CancelSandboxReset(SandboxResetTestContext(t.Context()), installation, 1) + err = deploymentExecution(t, w).CancelReset(SandboxResetTestContext(t.Context()), installation, 1) } if err != nil { t.Fatal(err) diff --git a/services/core/internal/store/sandbox_deployment_switch_test.go b/services/core/internal/store/sandbox_deployment_switch_test.go index 024cf4da6..a80bdb747 100644 --- a/services/core/internal/store/sandbox_deployment_switch_test.go +++ b/services/core/internal/store/sandbox_deployment_switch_test.go @@ -50,14 +50,14 @@ func TestSandboxResetClearsCustomE2BEndpoint(t *testing.T) { t.Fatal(err) } ctx := SandboxResetTestContext(t.Context()) - if err := w.StartSandboxReset(ctx, installation, SandboxResetRequest{ExpectedGeneration: configured.Generation, Clear: "force"}); err != nil { + if err := deploymentExecution(t, w).StartReset(ctx, installation, deployment.ResetRequest{ExpectedGeneration: configured.Generation, Clear: "force"}); err != nil { t.Fatal(err) } reset, err := deploymentService(t, s).View(ctx) if err != nil || reset.Reset == nil { t.Fatal("reset did not start", reset, err) } - generation, err := w.CompleteSandboxReset(ctx, installation, configured.Generation, reset.Reset.RequestedAt) + generation, err := deploymentExecution(t, w).CompleteReset(ctx, installation, configured.Generation, reset.Reset.RequestedAt) if err != nil || generation != configured.Generation+1 { t.Fatal("custom endpoint blocked reset completion", generation, err) } @@ -129,7 +129,7 @@ func TestSandboxDirectDeploymentOwnershipAndCleanSwitch(t *testing.T) { if err != nil || observed.Expired { t.Fatal("cloud inherited legacy node-less expiry", err) } - if err := w.StartSandboxReset(SandboxResetTestContext(t.Context()), id, SandboxResetRequest{Clear: "auto", ExpectedGeneration: 1}); err != nil { + if err := deploymentExecution(t, w).StartReset(SandboxResetTestContext(t.Context()), id, deployment.ResetRequest{Clear: "auto", ExpectedGeneration: 1}); err != nil { t.Fatal(err) } update := sandbox.Selection{DeploymentSpec: SandboxDeploymentTestSpec("docker"), Provider: "docker", ExpectedGeneration: 1} @@ -152,7 +152,7 @@ func TestSandboxDirectDeploymentOwnershipAndCleanSwitch(t *testing.T) { if err != nil || changed.Generation != 3 || changed.Mode != "nodes" || changed.Reset != nil || string(changed.Configuration) != "{}" || changed.Resources != (deployment.Resources{}) { t.Fatal("clean switch", changed, err) } - if err := w.CancelSandboxReset(SandboxResetTestContext(t.Context()), id, 1); !errors.Is(err, deployment.ErrConflict) { + if err := deploymentExecution(t, w).CancelReset(SandboxResetTestContext(t.Context()), id, 1); !errors.Is(err, deployment.ErrConflict) { t.Fatal("stale resume accepted", err) } if _, err := s.GetSession(t.Context(), tenant, session.ID); err != nil { @@ -186,7 +186,7 @@ func TestSandboxSwitchRetiresNodesAndEnrollment(t *testing.T) { if err != nil { t.Fatal(err) } - if err := w.StartSandboxReset(SandboxResetTestContext(t.Context()), id, SandboxResetRequest{Clear: "auto", ExpectedGeneration: 1}); err != nil { + if err := deploymentExecution(t, w).StartReset(SandboxResetTestContext(t.Context()), id, deployment.ResetRequest{Clear: "auto", ExpectedGeneration: 1}); err != nil { t.Fatal(err) } // AdmissionPaused rejects a valid enrollment without consuming it. Authentication @@ -216,13 +216,13 @@ func TestSandboxSwitchRetiresNodesAndEnrollment(t *testing.T) { if _, err := nodes.AuthenticateNode(t.Context(), node.NodeID, node.Credential); !errors.Is(err, deployment.ErrNodeCredential) { t.Fatal("old node credential survived", err) } - if err := w.StartSandboxReset(SandboxResetTestContext(t.Context()), id, SandboxResetRequest{Clear: "auto", ExpectedGeneration: 3}); err != nil { + if err := deploymentExecution(t, w).StartReset(SandboxResetTestContext(t.Context()), id, deployment.ResetRequest{Clear: "auto", ExpectedGeneration: 3}); err != nil { t.Fatal(err) } if _, err := resetAndSelect(t, w, id, 3, sandbox.Selection{DeploymentSpec: SandboxDeploymentTestSpec("docker"), Provider: "docker"}); err != nil { t.Fatal(err) } - if err := w.CancelSandboxReset(SandboxResetTestContext(t.Context()), id, 5); err != nil { + if err := deploymentExecution(t, w).CancelReset(SandboxResetTestContext(t.Context()), id, 5); err != nil { t.Fatal(err) } node.NodeID = uuid.NewString() @@ -260,7 +260,7 @@ func TestSandboxResetSerializesFreshDirectSessions(t *testing.T) { } }() } - if err := w.StartSandboxReset(SandboxResetTestContext(t.Context()), id, SandboxResetRequest{Clear: "auto", ExpectedGeneration: 1}); err != nil { + if err := deploymentExecution(t, w).StartReset(SandboxResetTestContext(t.Context()), id, deployment.ResetRequest{Clear: "auto", ExpectedGeneration: 1}); err != nil { t.Fatal(err) } wg.Wait() @@ -332,7 +332,7 @@ func TestSandboxSwitchPreservesReleasedAllocationAndItemHistory(t *testing.T) { if err := pool.QueryRow(t.Context(), "SELECT to_jsonb(a) FROM runtime_allocations a WHERE id=$1", owner.ID).Scan(&allocationBefore); err != nil { t.Fatal(err) } - if err := w.StartSandboxReset(SandboxResetTestContext(t.Context()), installation, SandboxResetRequest{Clear: "auto", ExpectedGeneration: 1}); err != nil { + if err := deploymentExecution(t, w).StartReset(SandboxResetTestContext(t.Context()), installation, deployment.ResetRequest{Clear: "auto", ExpectedGeneration: 1}); err != nil { t.Fatal(err) } if _, err := resetAndSelect(t, w, installation, 1, sandbox.Selection{DeploymentSpec: SandboxDeploymentTestSpec("docker"), Provider: "docker"}); err != nil { diff --git a/services/core/internal/store/sandbox_deployment_switch_worker_test.go b/services/core/internal/store/sandbox_deployment_switch_worker_test.go index 09f78cbc5..a1c0dd1cb 100644 --- a/services/core/internal/store/sandbox_deployment_switch_worker_test.go +++ b/services/core/internal/store/sandbox_deployment_switch_worker_test.go @@ -119,7 +119,7 @@ func TestSandboxWorkerSwitchesAndRecoversFailedActivation(t *testing.T) { } reset := func(generation uint64) deployment.View { t.Helper() - if _, err := w.StartSandboxReset(store.SandboxResetTestContext(t.Context()), store.SandboxResetRequest{ExpectedGeneration: generation, Clear: "force"}); err != nil { + if _, err := w.StartSandboxReset(store.SandboxResetTestContext(t.Context()), deployment.ResetRequest{ExpectedGeneration: generation, Clear: "force"}); err != nil { t.Fatal(err) } deadline := time.Now().Add(12 * time.Second) diff --git a/services/core/internal/store/sandbox_reset.go b/services/core/internal/store/sandbox_reset.go deleted file mode 100644 index fb6884e54..000000000 --- a/services/core/internal/store/sandbox_reset.go +++ /dev/null @@ -1,223 +0,0 @@ -package store - -import ( - "context" - "encoding/json" - "errors" - "math" - "time" - - "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/deployment" - - "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/adminaudit" - "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/db/sqlc" - "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/persistence/postgres/auditpg" - "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sessions" - "github.com/jackc/pgx/v5" - "github.com/jackc/pgx/v5/pgtype" -) - -var ErrSandboxResetAdmission = errors.New("hosted admission is paused for a sandbox reset") - -// SandboxInUseError rejects reset completion while hosted resources remain. -type SandboxInUseError struct{ Resources deployment.Resources } - -func (e *SandboxInUseError) Error() string { - return "hosted sandbox resources still belong to this deployment" -} -func (e *SandboxInUseError) Unwrap() error { return deployment.ErrConflict } - -type SandboxResetRequest struct { - ExpectedGeneration uint64 `json:"expected_generation" binding:"required" minimum:"0"` - Clear string `json:"clear" binding:"required" enums:"auto,force"` - DeadlineSeconds *int32 `json:"deadline_seconds,omitempty" minimum:"300" maximum:"86400"` -} - -func checkSandboxGeneration(d sqlc.RuntimeDeployment, installation string, generation uint64) error { - if uint64(d.Generation) != generation { - return &deployment.GenerationStaleError{CurrentGeneration: uint64(d.Generation)} - } - if !d.WebManaged || runtimeUUID(d.InstallationID) != installation { - return deployment.ErrConflict - } - return nil -} - -func (s *Store) resetTransaction(ctx context.Context, apply func(context.Context, *sqlc.Queries, sqlc.RuntimeDeployment) error) error { - if err := s.checkExecutionAuthority(); err != nil { - return err - } - return s.writer.Transaction(ctx, func(ctx context.Context, tx pgx.Tx) error { - q := s.queries.WithTx(tx) - d, err := q.LockRuntimeDeployment(ctx) - if err != nil { - return err - } - return apply(ctx, q, d) - }) -} - -// StartSandboxReset starts or escalates the reset. It returns no view: the -// caller reads the deployment after commit. -func (s *Store) StartSandboxReset(ctx context.Context, installation string, input SandboxResetRequest) error { - if input.Clear != "auto" && input.Clear != "force" { - return sessions.ErrInvalidInput - } - deadline := int32(3600) - if input.DeadlineSeconds != nil { - if input.Clear != "auto" || *input.DeadlineSeconds < 300 || *input.DeadlineSeconds > 86400 { - return sessions.ErrInvalidInput - } - deadline = *input.DeadlineSeconds - } - return s.resetTransaction(ctx, func(ctx context.Context, q *sqlc.Queries, d sqlc.RuntimeDeployment) error { - if err := checkSandboxGeneration(d, installation, input.ExpectedGeneration); err != nil { - return err - } - if d.ProviderKind == "" { - return deployment.ErrNotConfigured - } - if d.ResetClear.Valid { - if d.ResetClear.String != input.Clear { - if input.Clear != "force" { - return deployment.ErrResetInProgress - } - if err := q.ForceSandboxReset(ctx); err != nil { - return err - } - if err := auditpg.RecordDeploymentMutation(ctx, q, "reset_force", "sandbox_deployment", installation); err != nil { - return err - } - } - } else { - source, ok := adminaudit.FromContext(ctx) - if !ok { - return sessions.ErrInvalidInput - } - // The audit insert validates the provenance before this transaction - // can commit. Persist only the same non-secret typed source. - audit, err := json.Marshal(source) - if err != nil { - return err - } - if err := q.StartSandboxReset(ctx, sqlc.StartSandboxResetParams{Clear: pgtype.Text{String: input.Clear, Valid: true}, DeadlineSeconds: deadline, Audit: audit}); err != nil { - return err - } - return auditpg.RecordDeploymentMutation(ctx, q, "reset_start", "sandbox_deployment", installation) - } - return nil - }) -} - -// CancelSandboxReset cancels a running reset. It returns no view: the caller -// reads the deployment after commit. -func (s *Store) CancelSandboxReset(ctx context.Context, installation string, generation uint64) error { - return s.resetTransaction(ctx, func(ctx context.Context, q *sqlc.Queries, d sqlc.RuntimeDeployment) error { - if err := checkSandboxGeneration(d, installation, generation); err != nil { - return err - } - if d.ResetClear.Valid { - if err := q.CancelSandboxReset(ctx); err != nil { - return err - } - return auditpg.RecordDeploymentMutation(ctx, q, "reset_cancel", "sandbox_deployment", installation) - } - return nil - }) -} - -// AdvanceSandboxResetDeadline makes force escalation durable before selecting -// work. A restart cannot extend an administrator's original absolute deadline. -func (s *Store) AdvanceSandboxResetDeadline(ctx context.Context) error { - return s.resetTransaction(ctx, func(ctx context.Context, q *sqlc.Queries, d sqlc.RuntimeDeployment) error { - if d.ResetClear.String != "auto" || !d.ResetDeadlineAt.Valid || time.Now().Before(d.ResetDeadlineAt.Time) { - return nil - } - source, err := sandboxResetAudit(d) - if err != nil { - return err - } - if err := q.ForceSandboxReset(ctx); err != nil { - return err - } - return auditpg.RecordDeploymentMutation(adminaudit.WithSource(ctx, source), q, "reset_deadline", "sandbox_deployment", runtimeUUID(d.InstallationID)) - }) -} - -func sandboxResetAudit(d sqlc.RuntimeDeployment) (adminaudit.Source, error) { - var source adminaudit.Source - if !d.ResetClear.Valid || json.Unmarshal(d.ResetAudit, &source) != nil { - return source, sessions.ErrInvalidInput - } - return source, nil -} - -// CompleteSandboxReset must run after the manager has drained. It does not take -// Session locks: archive and admission always lock Session before deployment. -// It returns the committed, unconfigured generation. -func (s *Store) CompleteSandboxReset(ctx context.Context, installation string, generation uint64, requestedAt time.Time) (uint64, error) { - var committed uint64 - err := s.resetTransaction(ctx, func(ctx context.Context, q *sqlc.Queries, d sqlc.RuntimeDeployment) error { - if err := checkSandboxGeneration(d, installation, generation); err != nil { - return err - } - if !d.ResetClear.Valid || !d.ResetRequestedAt.Time.Equal(requestedAt) { - return deployment.ErrConflict - } - if d.Generation == math.MaxInt64 || d.OwnerEpoch == math.MaxInt64 { - return deployment.ErrConflict - } - resources, err := q.CountRuntimeDeploymentResources(ctx) - if err != nil { - return err - } - if resources.Allocations != 0 || resources.Pending != 0 { - return &SandboxInUseError{deployment.Resources{Allocations: resources.Allocations, Pending: resources.Pending}} - } - source, err := sandboxResetAudit(d) - if err != nil { - return err - } - if err := q.CompleteSandboxReset(ctx); err != nil { - return err - } - // The locked row advances by exactly one generation. - committed = uint64(d.Generation) + 1 - if err := q.RetireSandboxNodes(ctx); err != nil { - return err - } - if err := q.RetireSandboxEnrollments(ctx); err != nil { - return err - } - if err := q.ClearSandboxGenerations(ctx); err != nil { - return err - } - return auditpg.RecordDeploymentMutation(adminaudit.WithSource(ctx, source), q, "reset_complete", "sandbox_deployment", installation) - }) - if err != nil { - return 0, err - } - return committed, nil -} - -type SandboxResetSession struct{ SessionID, TenantID string } - -func (s *Store) ListSandboxResetSessions(ctx context.Context, after string, force bool) ([]SandboxResetSession, error) { - cursor := pgtype.UUID{Valid: true} - if after != "" { - var err error - cursor, err = parseID(after) - if err != nil { - return nil, err - } - } - rows, err := s.queries.ListSandboxResetSessions(ctx, sqlc.ListSandboxResetSessionsParams{AfterID: cursor, Force: force}) - if err != nil { - return nil, err - } - result := make([]SandboxResetSession, 0, len(rows)) - for _, row := range rows { - result = append(result, SandboxResetSession{SessionID: runtimeUUID(row.ID), TenantID: runtimeUUID(row.TenantID)}) - } - return result, nil -} diff --git a/services/core/internal/store/sandbox_reset_test.go b/services/core/internal/store/sandbox_reset_test.go index d734977e1..93812c870 100644 --- a/services/core/internal/store/sandbox_reset_test.go +++ b/services/core/internal/store/sandbox_reset_test.go @@ -22,14 +22,14 @@ func SandboxResetTestContext(ctx context.Context) context.Context { func resetAndSelect(t *testing.T, w *Store, installation string, generation uint64, input sandbox.Selection) (deployment.View, error) { t.Helper() ctx := SandboxResetTestContext(t.Context()) - if err := w.StartSandboxReset(ctx, installation, SandboxResetRequest{Clear: "auto", ExpectedGeneration: generation}); err != nil { + if err := deploymentExecution(t, w).StartReset(ctx, installation, deployment.ResetRequest{Clear: "auto", ExpectedGeneration: generation}); err != nil { return deployment.View{}, err } reset, err := deploymentService(t, w).View(ctx) if err != nil { return deployment.View{}, err } - empty, err := w.CompleteSandboxReset(ctx, installation, generation, reset.Reset.RequestedAt) + empty, err := deploymentExecution(t, w).CompleteReset(ctx, installation, generation, reset.Reset.RequestedAt) if err != nil { return deployment.View{}, err } @@ -57,9 +57,9 @@ func assertResetPartition(t *testing.T, view deployment.View) { // startReset starts or escalates the reset and returns the deployment view // read after it commits. -func startReset(t *testing.T, ctx context.Context, w *Store, installation string, input SandboxResetRequest) (deployment.View, error) { +func startReset(t *testing.T, ctx context.Context, w *Store, installation string, input deployment.ResetRequest) (deployment.View, error) { t.Helper() - if err := w.StartSandboxReset(ctx, installation, input); err != nil { + if err := deploymentExecution(t, w).StartReset(ctx, installation, input); err != nil { return deployment.View{}, err } return deploymentService(t, w).View(ctx) @@ -89,7 +89,7 @@ func TestSandboxResetAutoUsesStartedWorkAndLockedRecheck(t *testing.T) { case "suspended": runtimeSuspensionSQL(t, s.pool, `UPDATE runtime_allocations SET compute_phase='suspended', compute_retained_until=clock_timestamp()+interval '1 hour' WHERE id=$1`, owner.ID) } - reset, err := startReset(t, SandboxResetTestContext(t.Context()), w, installation, SandboxResetRequest{ExpectedGeneration: 1, Clear: "auto"}) + reset, err := startReset(t, SandboxResetTestContext(t.Context()), w, installation, deployment.ResetRequest{ExpectedGeneration: 1, Clear: "auto"}) if err != nil { t.Fatal(err) } @@ -98,7 +98,7 @@ func TestSandboxResetAutoUsesStartedWorkAndLockedRecheck(t *testing.T) { if (reset.Reset.Remaining.Busy == 1) != busy { t.Fatalf("wrong busy classification: %+v", reset.Reset.Remaining) } - page, err := w.ListSandboxResetSessions(t.Context(), "", false) + page, err := deploymentStore(w).ResetSessions(t.Context(), "", false) if err != nil || (len(page) == 0) != busy { t.Fatal("auto eligibility", page, err) } @@ -107,7 +107,7 @@ func TestSandboxResetAutoUsesStartedWorkAndLockedRecheck(t *testing.T) { if !errors.Is(err, ErrSandboxResetSessionBusy) { t.Fatal("auto cut active work", err) } - if err := w.StartSandboxReset(SandboxResetTestContext(t.Context()), installation, SandboxResetRequest{ExpectedGeneration: 1, Clear: "force"}); err != nil { + if err := deploymentExecution(t, w).StartReset(SandboxResetTestContext(t.Context()), installation, deployment.ResetRequest{ExpectedGeneration: 1, Clear: "force"}); err != nil { t.Fatal(err) } _, err = w.ArchiveSandboxResetSession(t.Context(), tenant, session.ID, 1, reset.Reset.RequestedAt) @@ -131,7 +131,7 @@ func TestSandboxResetCancellationABADeadlineAndGeneration(t *testing.T) { s, w, installation := managedArchiveFixture(t) tenant, session := managedArchiveSession(t, s, managerSessionInput(uuid.NewString())) ctx := SandboxResetTestContext(t.Context()) - first, err := startReset(t, ctx, w, installation, SandboxResetRequest{ExpectedGeneration: 1, Clear: "auto"}) + first, err := startReset(t, ctx, w, installation, deployment.ResetRequest{ExpectedGeneration: 1, Clear: "auto"}) if err != nil { t.Fatal(err) } @@ -139,14 +139,14 @@ func TestSandboxResetCancellationABADeadlineAndGeneration(t *testing.T) { t.Fatal("default deadline", first) } shorter := int32(300) - replay, err := startReset(t, ctx, w, installation, SandboxResetRequest{ExpectedGeneration: 1, Clear: "auto", DeadlineSeconds: &shorter}) + replay, err := startReset(t, ctx, w, installation, deployment.ResetRequest{ExpectedGeneration: 1, Clear: "auto", DeadlineSeconds: &shorter}) if err != nil || !replay.Reset.RequestedAt.Equal(first.Reset.RequestedAt) || !replay.Reset.DeadlineAt.Equal(*first.Reset.DeadlineAt) { t.Fatal("retry moved durable deadline", replay, err) } - if err = w.CancelSandboxReset(ctx, installation, 1); err != nil { + if err = deploymentExecution(t, w).CancelReset(ctx, installation, 1); err != nil { t.Fatal(err) } - second, err := startReset(t, ctx, w, installation, SandboxResetRequest{ExpectedGeneration: 1, Clear: "auto"}) + second, err := startReset(t, ctx, w, installation, deployment.ResetRequest{ExpectedGeneration: 1, Clear: "auto"}) if err != nil { t.Fatal(err) } @@ -154,26 +154,26 @@ func TestSandboxResetCancellationABADeadlineAndGeneration(t *testing.T) { t.Fatal("cancelled reset archived successor work", err) } runtimeSuspensionSQL(t, s.pool, `UPDATE runtime_deployment SET reset_deadline_at=clock_timestamp()-interval '1 second'`) - if err := w.AdvanceSandboxResetDeadline(t.Context()); err != nil { + if err := deploymentExecution(t, w).AdvanceResetDeadline(t.Context()); err != nil { t.Fatal(err) } forced, err := deploymentService(t, s).View(t.Context()) if err != nil || forced.Reset.Clear != "force" || forced.Reset.ForcedAt == nil || !forced.Reset.RequestedAt.Equal(second.Reset.RequestedAt) { t.Fatal("deadline not durable", forced, err) } - if err := w.StartSandboxReset(ctx, installation, SandboxResetRequest{ExpectedGeneration: 1, Clear: "auto"}); !errors.Is(err, deployment.ErrResetInProgress) { + if err := deploymentExecution(t, w).StartReset(ctx, installation, deployment.ResetRequest{ExpectedGeneration: 1, Clear: "auto"}); !errors.Is(err, deployment.ErrResetInProgress) { t.Fatal("force downgraded", err) } - if err := w.StartSandboxReset(ctx, installation, SandboxResetRequest{ExpectedGeneration: 0, Clear: "force"}); !errors.Is(err, deployment.ErrConflict) { + if err := deploymentExecution(t, w).StartReset(ctx, installation, deployment.ResetRequest{ExpectedGeneration: 0, Clear: "force"}); !errors.Is(err, deployment.ErrConflict) { t.Fatal("stale reset precedence", err) } if _, err := w.ArchiveSandboxResetSession(t.Context(), tenant, session.ID, 1, second.Reset.RequestedAt); err != nil { t.Fatal(err) } - if _, err := w.CompleteSandboxReset(ctx, installation, 1, first.Reset.RequestedAt); !errors.Is(err, deployment.ErrConflict) { + if _, err := deploymentExecution(t, w).CompleteReset(ctx, installation, 1, first.Reset.RequestedAt); !errors.Is(err, deployment.ErrConflict) { t.Fatal("cancelled reset finalized successor", err) } - committed, err := w.CompleteSandboxReset(ctx, installation, 1, second.Reset.RequestedAt) + committed, err := deploymentExecution(t, w).CompleteReset(ctx, installation, 1, second.Reset.RequestedAt) if err != nil || committed != 2 { t.Fatal("reset commit", committed, err) } @@ -181,10 +181,10 @@ func TestSandboxResetCancellationABADeadlineAndGeneration(t *testing.T) { if err != nil || empty.Provider != "" || empty.Generation != 2 || empty.Reset != nil || empty.InstallationID != installation || empty.Configuration != nil || empty.Specification != nil { t.Fatal("reset commit", empty, err) } - if err := w.CancelSandboxReset(ctx, installation, 1); !errors.Is(err, deployment.ErrConflict) { + if err := deploymentExecution(t, w).CancelReset(ctx, installation, 1); !errors.Is(err, deployment.ErrConflict) { t.Fatal("stale cancel", err) } - if err := w.CancelSandboxReset(ctx, installation, 2); err != nil { + if err := deploymentExecution(t, w).CancelReset(ctx, installation, 2); err != nil { t.Fatal("cancel without reset", err) } } @@ -193,11 +193,11 @@ func TestSandboxResetAutoRechecksTurnStartedAfterListing(t *testing.T) { s, w, installation := managedArchiveFixture(t) tenant, session := managedArchiveSession(t, s, managerSessionInput(uuid.NewString())) archiveAllocation(t, w, tenant, session, installation) - reset, err := startReset(t, SandboxResetTestContext(t.Context()), w, installation, SandboxResetRequest{ExpectedGeneration: 1, Clear: "auto"}) + reset, err := startReset(t, SandboxResetTestContext(t.Context()), w, installation, deployment.ResetRequest{ExpectedGeneration: 1, Clear: "auto"}) if err != nil { t.Fatal(err) } - page, err := w.ListSandboxResetSessions(t.Context(), "", false) + page, err := deploymentStore(w).ResetSessions(t.Context(), "", false) if err != nil || len(page) != 1 { t.Fatal(page, err) } @@ -216,7 +216,7 @@ func TestSandboxResetAutoRechecksTurnStartedAfterListing(t *testing.T) { if !reflect.DeepEqual(before, after) { t.Fatal("rejected admission left provisional rows") } - if err := w.CancelSandboxReset(SandboxResetTestContext(t.Context()), installation, 1); err != nil { + if err := deploymentExecution(t, w).CancelReset(SandboxResetTestContext(t.Context()), installation, 1); err != nil { t.Fatal(err) } if _, err := s.CreateSession(t.Context(), tenant, managerSessionInput(uuid.NewString())); err != nil { @@ -228,19 +228,19 @@ func TestSandboxResetAuditFailureRollsBackPauseAndCompletion(t *testing.T) { s, w, installation := managedArchiveFixture(t) rejectAdminAuditInsert(t, s) rejected := adminaudit.WithSource(t.Context(), adminaudit.Source{CredentialID: "reset-fixture", ActorLabel: "operator", RequestID: rejectedAdminRequest, TraceID: "reset-trace"}) - if err := w.StartSandboxReset(rejected, installation, SandboxResetRequest{ExpectedGeneration: 1, Clear: "force"}); err == nil { + if err := deploymentExecution(t, w).StartReset(rejected, installation, deployment.ResetRequest{ExpectedGeneration: 1, Clear: "force"}); err == nil { t.Fatal("reset committed without audit") } view, err := deploymentService(t, s).View(t.Context()) if err != nil || view.Reset != nil || view.Generation != 1 { t.Fatal("failed audit left reset state", view, err) } - reset, err := startReset(t, SandboxResetTestContext(t.Context()), w, installation, SandboxResetRequest{ExpectedGeneration: 1, Clear: "force"}) + reset, err := startReset(t, SandboxResetTestContext(t.Context()), w, installation, deployment.ResetRequest{ExpectedGeneration: 1, Clear: "force"}) if err != nil { t.Fatal(err) } runtimeSuspensionSQL(t, s.pool, `UPDATE runtime_deployment SET reset_audit=jsonb_set(reset_audit,'{RequestID}',to_jsonb($1::text))`, rejectedAdminRequest) - if _, err := w.CompleteSandboxReset(t.Context(), installation, 1, reset.Reset.RequestedAt); err == nil { + if _, err := deploymentExecution(t, w).CompleteReset(t.Context(), installation, 1, reset.Reset.RequestedAt); err == nil { t.Fatal("completion committed without audit") } view, err = deploymentService(t, s).View(t.Context()) @@ -263,7 +263,7 @@ func TestSandboxResetSnapshotCountsOfflineOwnershipOnce(t *testing.T) { if err := s.DeleteSession(t.Context(), tenant, deleted.ID); err != nil { t.Fatal(err) } - reset, err := startReset(t, SandboxResetTestContext(t.Context()), w, process.InstallationID, SandboxResetRequest{ExpectedGeneration: 1, Clear: "auto"}) + reset, err := startReset(t, SandboxResetTestContext(t.Context()), w, process.InstallationID, deployment.ResetRequest{ExpectedGeneration: 1, Clear: "auto"}) if err != nil { t.Fatal(err) } @@ -301,7 +301,7 @@ func TestSandboxResetSnapshotCountsOfflineOwnershipOnce(t *testing.T) { t.Fatal("offline ownership projection", view.Reset.Remaining) } } - if _, err := w.CompleteSandboxReset(t.Context(), process.InstallationID, 1, reset.Reset.RequestedAt); err == nil { + if _, err := deploymentExecution(t, w).CompleteReset(t.Context(), process.InstallationID, 1, reset.Reset.RequestedAt); err == nil { t.Fatal("offline resources were treated as cleaned") } if err := deploymentService(t, s).RemoveNode(t.Context(), process.LocalNodeID); !errors.Is(err, deployment.ErrNodeInUse) { @@ -331,25 +331,25 @@ func TestSandboxResetPaginationSkipsBusyPrefixAndPreservesSelfHosted(t *testing. } } selfTenant, self := managedArchiveSession(t, s, environmentInput(uuid.NewString(), "self_hosted", "/workspace")) - reset, err := startReset(t, SandboxResetTestContext(t.Context()), w, installation, SandboxResetRequest{ExpectedGeneration: 1, Clear: "auto"}) + reset, err := startReset(t, SandboxResetTestContext(t.Context()), w, installation, deployment.ResetRequest{ExpectedGeneration: 1, Clear: "auto"}) if err != nil { t.Fatal(err) } - page, err := w.ListSandboxResetSessions(t.Context(), "", false) + page, err := deploymentStore(w).ResetSessions(t.Context(), "", false) if err != nil || len(page) != 1 || page[0].SessionID != "00000000-0000-4000-8000-000000000035" { t.Fatal("busy prefix starved idle work", page, err) } if _, err := w.ArchiveSandboxResetSession(t.Context(), page[0].TenantID, page[0].SessionID, 1, reset.Reset.RequestedAt); err != nil { t.Fatal(err) } - if err := w.StartSandboxReset(SandboxResetTestContext(t.Context()), installation, SandboxResetRequest{ExpectedGeneration: 1, Clear: "force"}); err != nil { + if err := deploymentExecution(t, w).StartReset(SandboxResetTestContext(t.Context()), installation, deployment.ResetRequest{ExpectedGeneration: 1, Clear: "force"}); err != nil { t.Fatal(err) } - first, err := w.ListSandboxResetSessions(t.Context(), "", true) + first, err := deploymentStore(w).ResetSessions(t.Context(), "", true) if err != nil || len(first) != 32 { t.Fatal(first, err) } - second, err := w.ListSandboxResetSessions(t.Context(), first[31].SessionID, true) + second, err := deploymentStore(w).ResetSessions(t.Context(), first[31].SessionID, true) if err != nil || len(second) != 2 { t.Fatal(second, err) } @@ -365,7 +365,7 @@ func TestSandboxResetPaginationSkipsBusyPrefixAndPreservesSelfHosted(t *testing. func TestSandboxResetOwnerRestartRetainsDeadlineAndProvenance(t *testing.T) { s, w, installation := managedArchiveFixture(t) tenant, session := managedArchiveSession(t, s, managerSessionInput(uuid.NewString())) - reset, err := startReset(t, SandboxResetTestContext(t.Context()), w, installation, SandboxResetRequest{ExpectedGeneration: 1, Clear: "auto"}) + reset, err := startReset(t, SandboxResetTestContext(t.Context()), w, installation, deployment.ResetRequest{ExpectedGeneration: 1, Clear: "auto"}) if err != nil { t.Fatal(err) } @@ -380,17 +380,17 @@ func TestSandboxResetOwnerRestartRetainsDeadlineAndProvenance(t *testing.T) { if err != nil || !current.Reset.RequestedAt.Equal(reset.Reset.RequestedAt) || !current.Reset.DeadlineAt.Equal(*reset.Reset.DeadlineAt) { t.Fatal("restart moved reset deadline", current, err) } - if err := w.CancelSandboxReset(SandboxResetTestContext(t.Context()), installation, 1); err == nil { + if err := deploymentExecution(t, w).CancelReset(SandboxResetTestContext(t.Context()), installation, 1); err == nil { t.Fatal("detached writer cancelled successor reset") } runtimeSuspensionSQL(t, s.pool, `UPDATE runtime_deployment SET reset_deadline_at=clock_timestamp()-interval '1 second'`) - if err := successor.AdvanceSandboxResetDeadline(t.Context()); err != nil { + if err := deploymentExecution(t, successor).AdvanceResetDeadline(t.Context()); err != nil { t.Fatal(err) } if _, err := successor.ArchiveSandboxResetSession(t.Context(), tenant, session.ID, 1, reset.Reset.RequestedAt); err != nil { t.Fatal(err) } - committed, err := successor.CompleteSandboxReset(t.Context(), installation, 1, reset.Reset.RequestedAt) + committed, err := deploymentExecution(t, successor).CompleteReset(t.Context(), installation, 1, reset.Reset.RequestedAt) if err != nil || committed != 2 { t.Fatal(committed, err) } diff --git a/services/core/internal/store/sandbox_specification_store_test.go b/services/core/internal/store/sandbox_specification_store_test.go index 829a92b92..089ed0d8e 100644 --- a/services/core/internal/store/sandbox_specification_store_test.go +++ b/services/core/internal/store/sandbox_specification_store_test.go @@ -90,7 +90,7 @@ func TestSandboxSpecificationBootstrapReadDoesNotConsumeEnrollment(t *testing.T) if _, err := nodes.NodeConfiguration(t.Context(), "", token, 0); !errors.Is(err, deployment.ErrNodeCredential) { t.Fatal("consumed enrollment still authorized bootstrap", err) } - if err := w.StartSandboxReset(SandboxResetTestContext(t.Context()), view.InstallationID, SandboxResetRequest{Clear: "auto", ExpectedGeneration: view.Generation}); err != nil { + if err := deploymentExecution(t, w).StartReset(SandboxResetTestContext(t.Context()), view.InstallationID, deployment.ResetRequest{Clear: "auto", ExpectedGeneration: view.Generation}); err != nil { t.Fatal(err) } config, err := nodes.NodeConfiguration(t.Context(), node.NodeID, node.Credential, 0) @@ -268,9 +268,10 @@ func TestSandboxSpecificationAllocationRaceWithMaintenance(t *testing.T) { results <- result{session, owner, err} }() } + resets := deploymentExecution(t, w) go func() { <-start - maintenance <- w.StartSandboxReset(SandboxResetTestContext(t.Context()), view.InstallationID, SandboxResetRequest{Clear: "auto", ExpectedGeneration: view.Generation}) + maintenance <- resets.StartReset(SandboxResetTestContext(t.Context()), view.InstallationID, deployment.ResetRequest{Clear: "auto", ExpectedGeneration: view.Generation}) }() close(start) if err := <-maintenance; err != nil { @@ -319,7 +320,7 @@ func TestNodeBoundToAnotherPublicURLGetsNoNewSandboxes(t *testing.T) { if _, err := s.CreateSession(t.Context(), uuid.NewString(), managerSessionInput(uuid.NewString())); !errors.Is(err, deployment.ErrNodeUnavailable) { t.Fatal("placed a new sandbox on a node bound to the old address", err) } - bindings, err := s.AddressBindings(t.Context()) + bindings, err := deploymentService(t, s).AddressBindings(t.Context()) if err != nil || bindings.Nodes != 1 || bindings.NodesOnOtherAddress != 1 || bindings.HostedSandboxes != 0 { t.Fatal(bindings, err) }