Move Environments and devices into sessions and sessionpg

services/core/IMPLEMENTATION.md

@@ -18,7 +18,7 @@ Shared vocabulary has one owner each, and domains use it rather than copy it. `i
 
 `internal/sessions` owns the Session vocabulary: Sessions, Turns, inputs, Environments and their provisioning failures, function calls, Item and Artifact reads, executor credentials, and the errors Session operations return, which `api` maps in `writeSessionsError`. It also owns the Session change vocabulary and its decisions: the public changes that report Turn and Session transitions, what a Turn that ends settles, measured Turn usage and the Session activity a change reports. Sequences that several Session operations share are `sessions` procedures, each over the transaction interface it declares: `CancelTurn` and `CancelWork` cancel work, `TrackInputActivity` reports the input activity a write changes, `FailEnvironment` and `TerminateEnvironment` settle an Environment that failed or whose managed compute ended, `CreateEnvironmentDevice` creates and binds a hosted Environment's dedicated Runtime device, `CheckComputeAdmission`, `CheckFileWriteGate` and `CheckInputStart` gate new work, `AdmitFunctionResult` saves a submitted function result for its Turn's call, and `LoadRequiredActions` lists the calls a Turn waits on. `AppendTurnEvents` records a batch of a Turn's execution observations, which `NewJournalBatch` validates, and `AppendTurnEvent` records its terminal outcome; both project the new entries. `ProjectSource` projects one journal entry or admitted input into Items, the root Turn's usage and Subagents, and `ProjectInput` projects an admitted input. `ValidTransition` decides the Turn status transitions, and the Subagent projection decides each Subagent's identity, lifecycle, child Turn and child Item changes. `sessions.ExecutionOperations` runs the Session writes only the execution owner makes; each function-call operation loads the Turn and its call in one leased Session transaction, decides in `sessions`, then applies in `sessionpg`, and `AppendTurnEvents` records and projects a Turn's journal batch in one leased Session transaction. `internal/items` owns public Items: it projects observations, merges them into stored Items and builds the ordered events that report each Item change. Neither imports persistence. `internal/persistence/postgres/sessionpg` loads the facts those decisions read and applies them inside the caller's Session transaction, under the Session lock: it allocates event sequence positions, event IDs, Item positions, output indexes and Subagent IDs, writes the change journal, each Turn's execution journal and its counters, Items, Turn usage, Subagent bindings, child Turns and child Items, and Artifact settlement, and prunes the change journal. It decides nothing. `WithSession` runs a Session operation's transaction on the pool or the lease: it locks the Session, applies the operation, then prunes the change journal. `BindSession` binds one Session to a transaction's queries as a `SessionTx`, which implements the procedures' interfaces.
 
-Other adapters call only two `sessionpg` participants, on their own transaction's queries: `LockSession`, which orders a write against the Session's Turns, and `BindSession`, which runs a Session procedure, such as `CreateEnvironmentDevice`, in that transaction. The other `sessionpg` exports serve `store` until the Session operations leave it. `tests/fixtures` is test tooling that composes procedures over a pooled `WithSession`, which is why `WithSession` and `BindSession` stay exported after store goes; no production code may do this. `deploymentpg` may import `sessionpg`, and `sessionpg` never imports `deploymentpg`; deployment's placement fact-loading and reservation code sits below both, so either may call it and it imports neither. Likewise `deployment` may compose the `sessions` procedures, and `sessions` never imports `deployment`.
+Other adapters call only these `sessionpg` participants, on their own transaction's queries: `LockSession`, which orders a write against the Session's Turns; `BindSession`, which runs a Session procedure, such as `CreateEnvironmentDevice`, in that transaction; `LoadEnvironment` and `LoadSessionEnvironment`, which read the tenant's Environment of a live Session by Environment or by Session; and `CreateSessionEnvironment`, which stores the Environment of a Session being created. The other `sessionpg` exports serve `store` until the Session operations leave it. `tests/fixtures` is test tooling that composes procedures over a pooled `WithSession`, which is why `WithSession` and `BindSession` stay exported after store goes; no production code may do this. `deploymentpg` may import `sessionpg`, and `sessionpg` never imports `deploymentpg`; deployment's placement fact-loading and reservation code sits below both, so either may call it and it imports neither. Likewise `deployment` may compose the `sessions` procedures, and `sessions` never imports `deployment`.
 
 `store` is transitional. `store.New` builds a pooled Store, and `store.NewExecution(s, lease)` builds the execution writer on a lease it borrows. An execution-only operation on a pooled Store fails with `store.ErrExecutionAuthority`. New adapters do not copy that check: their execution repositories require a `*pgunit.Lease` at construction, their public repositories expose no execution operation, and the check goes away with `store`.
 
@@ -34,7 +34,7 @@ Domain owners, each with its PostgreSQL adapter under `internal/persistence/post
 - `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, 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. Function calls and their application receipts and the Turn execution journal run through `sessions.ExecutionOperations` on `sessionpg.NewExecution(lease)`, which the Worker receives as `execution.Owner.Sessions`.
+- `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), Session Artifacts (their reads, deletion and the staging of a Turn's export), Environments (their reads, the initialization list and the frozen setup and initial files, which `sessionpg.Store` opens with the credential key), devices (their reads, creation, revocation, heartbeats and Runtime enrollment) and executor credential authentication (`sessions.ExecutorCredentialReader`). `cmd/server` builds one `sessionpg.Store` with the credential key and wires the Service and the Store into `execution.Dispatcher.Sessions` and `SessionsReader`, into the api fields, into Runtime enrollment and into the daemon gateway; the Worker stages Artifacts through the Service. Function calls and their application receipts, the Turn execution journal, Environment initialization, connection observations and their reconciliation, and Session device binding run through `sessions.ExecutionOperations` on `sessionpg.NewExecution(lease)`, which the Worker receives as `execution.Owner.Sessions`.
 
 ## Request handling
 

services/core/cmd/device/main.go

@@ -17,8 +17,10 @@ import (
 
 	"github.com/MiniMax-AI/OpenAgentCore/internal/obs/log"
 	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/databaseurl"
+	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/persistence/postgres/pgunit"
+	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/persistence/postgres/sessionpg"
 	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/runtimedevice"
-	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/store"
+	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sessions"
 )
 
 func main() {
@@ -54,7 +56,12 @@ func run() error {
 		return errors.New("invalid execution database configuration")
 	}
 	defer pool.Close()
-	s := store.New(pool)
+	// Device provisioning opens no frozen Session data, so it needs no
+	// credential key.
+	s, err := sessions.NewService(sessionpg.New(pgunit.NewPool(pool), nil))
+	if err != nil {
+		return err
+	}
 	if *revoke != "" {
 		return s.RevokeDevice(ctx, *tenant, *revoke)
 	}

services/core/cmd/server/daemon_bootstrap_test.go

@@ -18,7 +18,7 @@ type bootstrapCredentialStore struct {
 }
 
 func (s bootstrapCredentialStore) GetDeviceCredential(context.Context, string) (runtimedevice.Credential, bool, error) {
-	return runtimedevice.Credential{ID: "runtime", Type: runtimegateway.RuntimeTypeAgentDaemon,
+	return runtimedevice.Credential{ID: "runtime", Type: runtimedevice.RuntimeTypeAgentDaemon,
 		CredentialHash: runtimedevice.HashCredential("synthetic-token"), RuntimeNodeID: s.nodeID, RuntimeAllocationID: s.allocationID}, true, nil
 }
 

services/core/cmd/server/executor_connections.go

@@ -5,16 +5,16 @@ import (
 
 	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/runtimeenrollment"
 	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/runtimegateway"
-	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/store"
+	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sessions"
 )
 
 // executorConnections observes enrolled executors through the Runtime gateway.
 // registry is nil when this Core has no gateway; no executor is then connected.
 type executorConnections struct {
-	store    *store.Store
+	sessions sessions.Reader
 	registry *runtimegateway.Registry
 }
 
 func (c executorConnections) ExecutorConnected(ctx context.Context, environment, digest string) (bool, error) {
-	return runtimeenrollment.RuntimeConnected(ctx, c.store, c.registry, environment, digest)
+	return runtimeenrollment.RuntimeConnected(ctx, c.sessions, c.registry, environment, digest)
 }

services/core/cmd/server/http_routes_test.go

@@ -103,7 +103,7 @@ func daemonComposition(t testing.TB) http.Handler {
 		Subagents:       struct{ api.Subagents }{},
 		Artifacts:       struct{ api.Artifacts }{},
 		ArtifactsReader: struct{ api.ArtifactsReader }{},
-		SessionAdmin:    struct{ api.SessionAdmin }{}, Environments: struct{ api.Environments }{}, ExecutorConnections: struct{ api.ExecutorConnections }{},
+		SessionAdmin:    struct{ api.SessionAdmin }{}, Environments: struct{ api.Environments }{}, EnvironmentsReader: struct{ api.EnvironmentsReader }{}, ExecutorConnections: struct{ api.ExecutorConnections }{},
 		Admin: struct{ api.Admin }{}, AdminAudit: struct{ api.AdminAudit }{}, WriteAudit: struct{ api.WriteAudit }{}, Metrics: struct{ api.Metrics }{},
 		RuntimeObservations: struct{ api.RuntimeObservations }{}, RuntimeHistory: struct{ api.RuntimeHistory }{},
 		Execution: &api.Execution{

services/core/cmd/server/main.go

@@ -164,7 +164,7 @@ func run() error {
 	if err != nil {
 		return err
 	}
-	sessionStore := sessionpg.New(units)
+	sessionStore := sessionpg.New(units, credentialKey)
 	sessionService, err := sessions.NewService(sessionStore)
 	if err != nil {
 		return err
@@ -248,7 +248,7 @@ func run() error {
 	if err := api.ValidateCredentialSeparation(ctx, keyAdmin, projectStore); err != nil {
 		return err
 	}
-	historyResolver, err := historystoreresolver.NewResolver(executionStore)
+	historyResolver, err := historystoreresolver.NewResolver(sessionStore)
 	if err != nil {
 		return err
 	}
@@ -265,7 +265,7 @@ func run() error {
 		if err != nil {
 			return err
 		}
-		daemonHandler, registry, err = runtime.NewGateway(executionStore, executorURL)
+		daemonHandler, registry, err = runtime.NewGateway(sessionStore, sessionService, executionStore, executorURL)
 		if err != nil {
 			return err
 		}
@@ -392,7 +392,7 @@ func run() error {
 		Artifacts:       sessionService,
 		ArtifactsReader: sessionStore,
 		SessionAdmin:    executionStore,
-		Environments:    executionStore, ExecutorConnections: executorConnections{store: executionStore, registry: registry},
+		Environments:    executionStore, EnvironmentsReader: sessionStore, ExecutorConnections: executorConnections{sessions: sessionStore, registry: registry},
 		Admin: executionStore, AdminAudit: auditStore, WriteAudit: auditStore, Metrics: metrics,
 		RuntimeObservations: observationService, RuntimeHistory: historyService,
 	}
@@ -421,8 +421,8 @@ func run() error {
 	}
 	if daemonHandler != nil {
 		routes := daemonRoutes{gateway: daemonHandler,
-			enrollment: runtimeenrollment.EnrollmentHandler(executionStore),
-			connection: runtimeenrollment.ConnectionHandler(executionStore, registry)}
+			enrollment: runtimeenrollment.EnrollmentHandler(sessionService),
+			connection: runtimeenrollment.ConnectionHandler(sessionStore, registry)}
 		if managedNodes != nil {
 			routes.nodeConnect = managedNodes.hub
 		}

services/core/internal/api/dependencies.go

@@ -52,6 +52,7 @@ type Dependencies struct {
 	ArtifactsReader      ArtifactsReader
 	SessionAdmin         SessionAdmin
 	Environments         Environments
+	EnvironmentsReader   EnvironmentsReader
 	ExecutorConnections  ExecutorConnections
 	Admin                Admin
 	AdminAudit           AdminAudit
@@ -139,7 +140,7 @@ func (d Dependencies) validate() error {
 		field{"Subagents", d.Subagents},
 		field{"Artifacts", d.Artifacts},
 		field{"ArtifactsReader", d.ArtifactsReader},
-		field{"SessionAdmin", d.SessionAdmin}, field{"Environments", d.Environments},
+		field{"SessionAdmin", d.SessionAdmin}, field{"Environments", d.Environments}, field{"EnvironmentsReader", d.EnvironmentsReader},
 		field{"ExecutorConnections", d.ExecutorConnections}, field{"Admin", d.Admin}, field{"AdminAudit", d.AdminAudit}, field{"WriteAudit", d.WriteAudit},
 		field{"Metrics", d.Metrics}, field{"RuntimeObservations", d.RuntimeObservations}, field{"RuntimeHistory", d.RuntimeHistory},
 	); err != nil {

services/core/internal/api/dependencies_test.go

@@ -38,6 +38,7 @@ type testFakes struct {
 	artifactsReader        *fakeArtifactsReader
 	sessionAdmin           *fakeSessionAdmin
 	environments           *fakeEnvironments
+	environmentsReader     *fakeEnvironmentsReader
 	executorConnections    *fakeExecutorConnections
 	admin                  *fakeAdmin
 	adminAudit             *fakeAdminAudit
@@ -82,7 +83,7 @@ func testDependencies(t testing.TB) (Dependencies, *testFakes) {
 		subagents:       &fakeSubagents{t: t},
 		artifacts:       &fakeArtifacts{t: t},
 		artifactsReader: &fakeArtifactsReader{t: t},
-		sessionAdmin:    &fakeSessionAdmin{t: t}, environments: &fakeEnvironments{t: t}, executorConnections: &fakeExecutorConnections{t: t},
+		sessionAdmin:    &fakeSessionAdmin{t: t}, environments: &fakeEnvironments{t: t}, environmentsReader: &fakeEnvironmentsReader{t: t}, executorConnections: &fakeExecutorConnections{t: t},
 		admin: &fakeAdmin{t: t}, adminAudit: &fakeAdminAudit{t: t}, writeAudit: &fakeWriteAudit{t: t}, metrics: &fakeMetrics{t: t},
 		runtimeObservations: &fakeRuntimeObservations{t: t}, runtimeHistory: &fakeRuntimeHistory{t: t}, installationBindings: &fakeInstallationBindings{t: t},
 		sessionAdmission: &fakeSessionAdmission{t: t},
@@ -110,7 +111,7 @@ func testDependencies(t testing.TB) (Dependencies, *testFakes) {
 		Artifacts:       f.artifacts,
 		ArtifactsReader: f.artifactsReader,
 		SessionAdmin:    f.sessionAdmin,
-		Environments:    f.environments, ExecutorConnections: f.executorConnections, Admin: f.admin, AdminAudit: f.adminAudit, WriteAudit: f.writeAudit,
+		Environments:    f.environments, EnvironmentsReader: f.environmentsReader, ExecutorConnections: f.executorConnections, Admin: f.admin, AdminAudit: f.adminAudit, WriteAudit: f.writeAudit,
 		Metrics: f.metrics, RuntimeObservations: f.runtimeObservations, RuntimeHistory: f.runtimeHistory,
 	}, f
 }

services/core/internal/api/environment_files.go

@@ -38,7 +38,7 @@ type EnvironmentWorkspaces interface {
 // @Failure 400,401,404,500,503 {object} v1.ErrorResponse
 // @Router /agents/environments/{environment_id}/files [get]
 func (h *Handler) listEnvironmentFiles(w http.ResponseWriter, r *http.Request) {
-	environment, err := h.Environments.GetEnvironment(r.Context(), tenantID(r), chi.URLParam(r, "environment_id"))
+	environment, err := h.EnvironmentsReader.GetEnvironment(r.Context(), tenantID(r), chi.URLParam(r, "environment_id"))
 	if err != nil {
 		writeStoreError(w, r, err)
 		return

services/core/internal/api/environment_files_create.go

@@ -40,7 +40,7 @@ func (h *Handler) createEnvironmentFile(w http.ResponseWriter, r *http.Request)
 	if !ok {
 		return
 	}
-	environment, err := h.Environments.GetEnvironment(r.Context(), tenantID(r), chi.URLParam(r, "environment_id"))
+	environment, err := h.EnvironmentsReader.GetEnvironment(r.Context(), tenantID(r), chi.URLParam(r, "environment_id"))
 	if err != nil {
 		writeStoreError(w, r, err)
 		return

services/core/internal/api/environment_files_create_test.go

@@ -47,7 +47,7 @@ func environmentFileCreateHandler(t *testing.T, configure ...func(*Dependencies,
 		APIKey{OrganizationID: "org", ProjectID: "project", SubjectKind: "user", SubjectID: "caller", TokenSHA256: runtimedevice.HashCredential("files-key"), TenantID: f.environment.TenantID},
 		APIKey{OrganizationID: "org", ProjectID: "other", SubjectKind: "user", SubjectID: "other", TokenSHA256: runtimedevice.HashCredential("other-key"), TenantID: uuid.NewString()},
 	).ResolveAPIKey
-	fakes.environments.getEnvironment = f.GetEnvironment
+	fakes.environmentsReader.getEnvironment = f.GetEnvironment
 	deps.Execution = fakes.execution()
 	fakes.workspaces.readEnvironmentDirectory, fakes.workspaces.writeEnvironmentFile = f.ReadEnvironmentDirectory, f.WriteEnvironmentFile
 	for _, c := range configure {

services/core/internal/api/environment_files_test.go

@@ -79,7 +79,7 @@ func environmentFilesHandler(t *testing.T, enabled bool, configure ...func(*Depe
 	deps, fakes := testDependencies(t)
 	deps.Engine = "fake_alpha"
 	fakes.projectsReader.resolveAPIKey = projectKeys(t, keys...).ResolveAPIKey
-	fakes.environments.getEnvironment = f.GetEnvironment
+	fakes.environmentsReader.getEnvironment = f.GetEnvironment
 	if enabled {
 		deps.Execution = fakes.execution()
 		fakes.workspaces.readEnvironmentDirectory = f.ReadEnvironmentDirectory