Move executor credentials, installations and file writes into sessions and sessionpg

contracts/agents-api/core.openapi.yaml

@@ -4258,6 +4258,14 @@ paths:
           description: Conflict
           schema:
             $ref: '#/definitions/api.CoreErrorResponse'
+        "500":
+          description: Internal Server Error
+          schema:
+            $ref: '#/definitions/api.CoreErrorResponse'
+        "503":
+          description: Service Unavailable
+          schema:
+            $ref: '#/definitions/api.CoreErrorResponse'
       security:
         - DeploymentAdminAuth: []
       summary: Get a self_hosted Session's installation commands

contracts/agents-api/environment-executor-credentials.md

@@ -33,7 +33,7 @@ The installer calls these machine routes on Core:
 | `POST /api/v1/agent-daemon/installation` | Grant | The frozen binding: `version`, `protocol_version`, `environment_id`, `remote_url`, `workspace_directory`, `harness` |
 | `POST /api/v1/agent-daemon/installation/claim` | Grant | `{"executor_token":"SECRET"}`; 204 |
 
-An invalid or expired grant returns 401 `installation_authorization_invalid`. Without matching installers the grant routes return 503 `installation_unavailable`. A malformed secret returns 400. Artifact routes carry no credential, and the grant is sent only to Core, never to an artifact host.
+An invalid or expired grant returns 401 `installation_authorization_invalid`. Without matching installers the grant routes return 503 `installation_unavailable`. Core signs each grant with the installation's [`secrets/credential.key`](../../docs/configuration.md#installation-directory); without a configured key, the Session responses and Core-key read above and the grant routes return 503 `credential_storage_unavailable`. A malformed secret returns 400. Artifact routes carry no credential, and the grant is sent only to Core, never to an artifact host.
 
 ## Core-key routes
 

contracts/agents-api/openapi.yaml

@@ -3270,6 +3270,10 @@ paths:
           description: Internal Server Error
           schema:
             $ref: '#/definitions/v1.ErrorResponse'
+        "503":
+          description: Service Unavailable
+          schema:
+            $ref: '#/definitions/v1.ErrorResponse'
       security:
       - BearerAuth: []
       summary: Retrieve an execution Session
@@ -3330,6 +3334,10 @@ paths:
           description: Internal Server Error
           schema:
             $ref: '#/definitions/v1.ErrorResponse'
+        "503":
+          description: Service Unavailable
+          schema:
+            $ref: '#/definitions/v1.ErrorResponse'
       security:
       - BearerAuth: []
       summary: Update execution Session metadata

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 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`.
+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`, which reads the tenant's Environment of a live 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), 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`.
+- `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), executor credentials (authentication and a Project's credential state through `sessions.ExecutorCredentialReader`, and issuance, rotation and revocation; the Core-key Project operations record their audit entry in the same transaction) and native installation authorization and claims, whose tokens `sessionpg.Store` signs with the credential key. `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; `cmd/environment-key` builds one without the key for its credential commands. Function calls and their application receipts, the Turn execution journal, Environment initialization, connection observations and their reconciliation, Session device binding, and file-write reservation and settlement run through `sessions.ExecutionOperations` on `sessionpg.NewExecution(lease)`, which the Worker receives as `execution.Owner.Sessions`.
 
 ## Request handling
 

services/core/cmd/environment-key/main.go

@@ -13,8 +13,9 @@ 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/identity"
+	"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/sessions"
-	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/store"
 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5/pgxpool"
 )
@@ -108,15 +109,19 @@ func run() error {
 		return errors.New("invalid execution database configuration")
 	}
 	defer pool.Close()
-	s := store.New(pool)
+	// Executor credentials need no credential key.
+	credentials, err := sessions.NewService(sessionpg.New(pgunit.NewPool(pool), nil))
+	if err != nil {
+		return err
+	}
 	if options.revoke {
-		return credentialOperationError(s.RevokeExecutorCredential(ctx, options.principal, options.keyID))
+		return credentialOperationError(credentials.RevokeExecutorCredential(ctx, options.principal, options.keyID))
 	}
 	var credential sessions.IssuedExecutorCredential
 	if options.rotate {
-		credential, err = s.RotateExecutorCredential(ctx, options.principal, options.keyID)
+		credential, err = credentials.RotateExecutorCredential(ctx, options.principal, options.keyID)
 	} else {
-		credential, err = s.IssueExecutorCredential(ctx, options.principal, options.keyID, options.environment)
+		credential, err = credentials.IssueExecutorCredential(ctx, options.principal, options.keyID, options.environment)
 	}
 	if err != nil {
 		return credentialOperationError(err)

services/core/cmd/server/main.go

@@ -392,7 +392,7 @@ func run() error {
 		Artifacts:       sessionService,
 		ArtifactsReader: sessionStore,
 		SessionAdmin:    executionStore,
-		Environments:    executionStore, EnvironmentsReader: sessionStore, ExecutorConnections: executorConnections{sessions: sessionStore, registry: registry},
+		Environments:    sessionService, EnvironmentsReader: sessionStore, ExecutorConnections: executorConnections{sessions: sessionStore, registry: registry},
 		Admin: executionStore, AdminAudit: auditStore, WriteAudit: auditStore, Metrics: metrics,
 		RuntimeObservations: observationService, RuntimeHistory: historyService,
 	}

services/core/internal/api/environment_executor_management.go

@@ -66,9 +66,9 @@ func (h *Handler) listExecutorCredentials(w http.ResponseWriter, r *http.Request
 	if !ok {
 		return
 	}
-	state, err := h.Environments.ProjectExecutorCredentialState(r.Context(), binding.Principal, chi.URLParam(r, "environment_id"))
+	state, err := h.EnvironmentsReader.ProjectExecutorCredentialState(r.Context(), binding.Principal, chi.URLParam(r, "environment_id"))
 	if err != nil {
-		writeStoreError(w, r, err)
+		writeSessionsError(w, r, err)
 		return
 	}
 	connection := ExecutorConnection{Status: "never_enrolled"}
@@ -111,7 +111,7 @@ func (h *Handler) issueExecutorCredential(w http.ResponseWriter, r *http.Request
 	var input EnvironmentExecutorCredentialRequest
 	var fields map[string]json.RawMessage
 	if decodeInputObject(raw, &input, "key_id", "rotate") != nil || json.Unmarshal(raw, &fields) != nil || bytes.Equal(bytes.TrimSpace(fields["rotate"]), []byte("null")) || !executorManagementID(input.KeyID) {
-		writeStoreError(w, r, sessions.ErrInvalidInput)
+		writeSessionsError(w, r, sessions.ErrInvalidInput)
 		return
 	}
 	binding, ok := h.adminProjectScope(w, r)
@@ -120,7 +120,7 @@ func (h *Handler) issueExecutorCredential(w http.ResponseWriter, r *http.Request
 	}
 	credential, err := h.Environments.IssueProjectExecutorCredential(r.Context(), binding.Principal, chi.URLParam(r, "environment_id"), input.KeyID, input.Rotate)
 	if err != nil {
-		writeStoreError(w, r, err)
+		writeSessionsError(w, r, err)
 		return
 	}
 	writeJSON(w, http.StatusCreated, credential)
@@ -143,11 +143,11 @@ func (h *Handler) revokeExecutorCredential(w http.ResponseWriter, r *http.Reques
 	}
 	keyID := chi.URLParam(r, "key_id")
 	if !executorManagementID(keyID) {
-		writeStoreError(w, r, sessions.ErrNotFound)
+		writeSessionsError(w, r, sessions.ErrNotFound)
 		return
 	}
 	if err := h.Environments.RevokeProjectExecutorCredential(r.Context(), binding.Principal, chi.URLParam(r, "environment_id"), keyID); err != nil {
-		writeStoreError(w, r, err)
+		writeSessionsError(w, r, err)
 		return
 	}
 	w.Header().Set("Cache-Control", "no-store")

services/core/internal/api/environment_executor_management_test.go

@@ -54,7 +54,7 @@ func (f *executorManagementFixture) RevokeProjectExecutorCredential(ctx context.
 func executorManagementHandler(t *testing.T, key APIKey, f *executorManagementFixture, connected func(context.Context, string, string) (bool, error)) http.Handler {
 	t.Helper()
 	deps, fakes := managementFakes(t, key)
-	fakes.environments.projectExecutorCredentialState = f.ProjectExecutorCredentialState
+	fakes.environmentsReader.projectExecutorCredentialState = f.ProjectExecutorCredentialState
 	fakes.environments.issueProjectExecutorCredential = f.IssueProjectExecutorCredential
 	fakes.environments.revokeProjectExecutorCredential = f.RevokeProjectExecutorCredential
 	fakes.executorConnections.executorConnected = connected