Share Session transaction procedures through sessions and sessionpg

services/core/IMPLEMENTATION.md

@@ -16,7 +16,9 @@ Two errors are shared across domains, each with one `api` helper: `textvalue.Err
 
 Shared vocabulary has one owner each, and domains use it rather than copy it. `internal/environmentconfig` owns Environment setup, Skills, Plugins and initial files with their validation and public metadata; `Setup.Validate` checks requested configuration, where a Skill may be an unresolved reference, and `Setup.ValidateInstalled` checks frozen, installable configuration. `internal/skills` owns `ParseVersion`, the canonical positive decimal Skill version. `internal/metadata` owns the metadata rules: `Validate` for the pair, key and value limits and U+0000, `ValidateStorable` for U+0000 alone, and `Encode` with its 64 KiB bound. `internal/jsonobject` owns `Normalize`, the stable encoding of stored JSON objects that snapshots and retry identities compare. These packages import no persistence.
 
-`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. `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 and output indexes, writes the journal, Items, Turn usage and Artifact settlement, and prunes the journal. It decides nothing.
+`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, and `CheckComputeAdmission`, `CheckFileWriteGate` and `CheckInputStart` gate new work. `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 and output indexes, writes the journal, Items, Turn usage and Artifact settlement, and prunes the 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 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. `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`.
 

services/core/internal/db/queries/environment_connections.sql

@@ -13,9 +13,15 @@ UPDATE environment_connections SET revision = $2 WHERE environment_id = $1;
 UPDATE environments SET status = $2 WHERE id = $1;
 
 -- name: RecordEnvironmentFailure :one
-UPDATE environments SET status = 'failed', failure_reason = $2, failure_detail = $3, failed_at = clock_timestamp()
-WHERE id = $1 AND status NOT IN ('failed', 'expired')
-RETURNING failed_at;
+UPDATE environments e SET status = 'failed', failure_reason = sqlc.arg(failure_reason), failure_detail = sqlc.arg(failure_detail), failed_at = clock_timestamp()
+WHERE e.id = sqlc.arg(id) AND e.session_id = sqlc.arg(session_id) AND e.status NOT IN ('failed', 'expired')
+AND EXISTS (SELECT 1 FROM sessions s WHERE s.id = e.session_id AND s.tenant_id = sqlc.arg(tenant_id))
+RETURNING e.failed_at;
+
+-- name: ExpireSessionEnvironment :execrows
+UPDATE environments e SET status = 'expired'
+WHERE e.id = sqlc.arg(id) AND e.session_id = sqlc.arg(session_id)
+AND EXISTS (SELECT 1 FROM sessions s WHERE s.id = e.session_id AND s.tenant_id = sqlc.arg(tenant_id));
 
 -- name: DeleteEnvironmentConnection :exec
 DELETE FROM environment_connections WHERE environment_id = $1;

services/core/internal/db/queries/environment_input_expiry.sql

@@ -1,5 +1,5 @@
 -- name: ListDueEnvironmentInputs :many
-SELECT r.id, r.session_id
+SELECT r.id, r.session_id, s.tenant_id
 FROM environment_input_reservations r
 JOIN sessions s ON s.id = r.session_id
 WHERE r.state = 'pending'

services/core/internal/db/queries/local_environment_devices.sql

@@ -2,7 +2,7 @@
 INSERT INTO devices (id, tenant_id, name, credential_hash, environment_id)
 SELECT sqlc.arg(id), s.tenant_id, sqlc.arg(name), sqlc.arg(credential_hash), e.id
 FROM environments e JOIN sessions s ON s.id = e.session_id
-WHERE s.tenant_id = sqlc.arg(tenant_id) AND e.id = sqlc.arg(environment_id)
+WHERE s.tenant_id = sqlc.arg(tenant_id) AND s.id = sqlc.arg(session_id) AND e.id = sqlc.arg(environment_id)
 AND s.deleted_at IS NULL AND s.configuration->'environment'->>'type' = 'openai_hosted'
 ON CONFLICT (environment_id) DO NOTHING
 RETURNING id;

services/core/internal/persistence/postgres/pgunit/pgunit.go

@@ -19,6 +19,17 @@ var (
 	snapshot  = pgx.TxOptions{IsoLevel: pgx.RepeatableRead, AccessMode: pgx.ReadOnly}
 )
 
+// Transactor runs read committed transactions: a Pool on pooled connections,
+// or the Lease on the execution connection.
+type Transactor interface {
+	Transaction(ctx context.Context, apply func(context.Context, pgx.Tx) error) error
+}
+
+var (
+	_ Transactor = (*Pool)(nil)
+	_ Transactor = (*Lease)(nil)
+)
+
 // Pool runs transactions on pooled connections. It grants no execution authority.
 type Pool struct{ pool *pgxpool.Pool }
 

services/core/internal/persistence/postgres/sessionpg/binding.go

@@ -0,0 +1,192 @@
+package sessionpg
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgtype"
+
+	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/db/sqlc"
+	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sessions"
+)
+
+// SessionTx is one Session inside its caller's transaction. It implements the
+// transaction interfaces of the sessions procedures by loading the facts they
+// read and applying what they decide, and decides nothing itself.
+//
+// The caller guarantees what the procedures assume: the queries are bound to
+// the caller's open transaction, the tenant owns the Session, the transaction
+// already holds the Session lock (LockSession), and the operation that owns the
+// transaction supplies its authority, such as the execution lease. A SessionTx
+// never begins or commits a transaction, never chooses a pool or the lease, and
+// is not used after the transaction callback it was bound in returns.
+type SessionTx struct {
+	q               *sqlc.Queries
+	tenant, session pgtype.UUID
+}
+
+var (
+	_ sessions.EnvironmentTerminationTx = (*SessionTx)(nil)
+	_ sessions.InputStartTx             = (*SessionTx)(nil)
+	_ sessions.ComputeAdmissionTx       = (*SessionTx)(nil)
+	_ sessions.EnvironmentDeviceTx      = (*SessionTx)(nil)
+)
+
+// BindSession binds the tenant's Session to the caller's transaction-bound
+// queries.
+func BindSession(q *sqlc.Queries, tenant, session pgtype.UUID) *SessionTx {
+	return &SessionTx{q: q, tenant: tenant, session: session}
+}
+
+func (t *SessionTx) LoadUsage(ctx context.Context) (json.RawMessage, error) {
+	return LoadUsage(ctx, t.q, t.session)
+}
+
+func (t *SessionTx) AppendChanges(ctx context.Context, changes ...sessions.SessionChange) error {
+	return AppendChanges(ctx, t.q, t.session, changes...)
+}
+
+func (t *SessionTx) LoadActiveTurn(ctx context.Context) (sessions.Turn, bool, error) {
+	row, err := t.q.GetActiveTurn(ctx, t.session)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return sessions.Turn{}, false, nil
+	}
+	if err != nil {
+		return sessions.Turn{}, false, err
+	}
+	return TurnFromRow(row), true, nil
+}
+
+func (t *SessionTx) LoadTurn(ctx context.Context, turn string) (sessions.Turn, error) {
+	id, err := parseID(turn)
+	if err != nil {
+		return sessions.Turn{}, err
+	}
+	row, err := t.q.SessionEventTurn(ctx, sqlc.SessionEventTurnParams{SessionID: t.session, ID: id})
+	if err != nil {
+		return sessions.Turn{}, err
+	}
+	return TurnFromRow(row), nil
+}
+
+func (t *SessionTx) RequestTurnCancel(ctx context.Context, turn string) error {
+	id, err := parseID(turn)
+	if err != nil {
+		return err
+	}
+	return t.q.RequestTurnCancel(ctx, sqlc.RequestTurnCancelParams{ID: id, SessionID: t.session})
+}
+
+func (t *SessionTx) LoadEnding(ctx context.Context, turn string) (sessions.Ending, error) {
+	id, err := parseID(turn)
+	if err != nil {
+		return sessions.Ending{}, err
+	}
+	return LoadEnding(ctx, t.q, t.session, id)
+}
+
+func (t *SessionTx) ApplyTurnEnd(ctx context.Context, turn string, end sessions.TurnEnd) error {
+	id, err := parseID(turn)
+	if err != nil {
+		return err
+	}
+	return ApplyTurnEnd(ctx, t.q, t.session, id, end)
+}
+
+func (t *SessionTx) CancelPendingInput(ctx context.Context) error {
+	return t.q.CancelSessionEnvironmentInput(ctx, t.session)
+}
+
+func (t *SessionTx) FailPendingInput(ctx context.Context) error {
+	return t.q.FailSessionEnvironmentInput(ctx, t.session)
+}
+
+func (t *SessionTx) LoadEnvironmentInput(ctx context.Context) (*sessions.EnvironmentInputState, error) {
+	return LoadEnvironmentInput(ctx, t.q, t.session)
+}
+
+func (t *SessionTx) LoadEnvironment(ctx context.Context) (sessions.Environment, error) {
+	row, err := t.q.GetSessionEnvironment(ctx, sqlc.GetSessionEnvironmentParams{TenantID: t.tenant, ID: t.session})
+	return EnvironmentFromRow(row.Environment, row.TenantID, row.Configuration, err)
+}
+
+// RecordEnvironmentFailure fails the Environment only while it is the
+// Session's and still live; otherwise it changes nothing and returns
+// pgx.ErrNoRows.
+func (t *SessionTx) RecordEnvironmentFailure(ctx context.Context, environment, reason string, detail *sessions.ProvisioningFailureDetail) (time.Time, error) {
+	id, err := parseID(environment)
+	if err != nil {
+		return time.Time{}, err
+	}
+	var rawDetail []byte
+	if detail != nil {
+		if rawDetail, err = json.Marshal(detail); err != nil {
+			return time.Time{}, err
+		}
+	}
+	failedAt, err := t.q.RecordEnvironmentFailure(ctx, sqlc.RecordEnvironmentFailureParams{
+		ID: id, SessionID: t.session, TenantID: t.tenant,
+		FailureReason: pgtype.Text{String: reason, Valid: true}, FailureDetail: rawDetail,
+	})
+	return failedAt.Time, err
+}
+
+// ExpireEnvironment expires the Environment only while it is the Session's;
+// another Environment is sessions.ErrNotFound and changes nothing.
+func (t *SessionTx) ExpireEnvironment(ctx context.Context, environment string) error {
+	id, err := parseID(environment)
+	if err != nil {
+		return err
+	}
+	count, err := t.q.ExpireSessionEnvironment(ctx, sqlc.ExpireSessionEnvironmentParams{ID: id, SessionID: t.session, TenantID: t.tenant})
+	if err == nil && count != 1 {
+		return sessions.ErrNotFound
+	}
+	return err
+}
+
+func (t *SessionTx) LoadBoundDevice(ctx context.Context) (bool, error) {
+	_, err := t.q.GetSessionDevice(ctx, sqlc.GetSessionDeviceParams{TenantID: t.tenant, ID: t.session})
+	if errors.Is(err, pgx.ErrNoRows) {
+		return false, nil
+	}
+	return err == nil, err
+}
+
+// InsertEnvironmentDevice inserts the device only for the Session's own hosted
+// Environment. The insert takes no row when the Environment already has a
+// device, including one a concurrent transaction inserted first, or cannot
+// take one: that is sessions.ErrDeviceBindingConflict.
+func (t *SessionTx) InsertEnvironmentDevice(ctx context.Context, device sessions.ExecutionDevice, credentialHash string) error {
+	id, err := parseID(device.ID)
+	if err != nil {
+		return err
+	}
+	environment, err := parseID(device.EnvironmentID)
+	if err != nil {
+		return err
+	}
+	inserted, err := t.q.CreateEnvironmentDevice(ctx, sqlc.CreateEnvironmentDeviceParams{
+		ID: id, TenantID: t.tenant, SessionID: t.session, EnvironmentID: environment,
+		Name: device.Name, CredentialHash: pgtype.Text{String: credentialHash, Valid: true},
+	})
+	if errors.Is(err, pgx.ErrNoRows) {
+		return sessions.ErrDeviceBindingConflict
+	}
+	if err != nil {
+		return err
+	}
+	_, err = t.q.BindSessionDevice(ctx, sqlc.BindSessionDeviceParams{TenantID: t.tenant, ID: t.session, ID_2: inserted})
+	return err
+}
+
+func (t *SessionTx) LoadComputeSuspension(ctx context.Context) (bool, error) {
+	return t.q.RuntimeComputeBlocksAdmission(ctx, t.session)
+}
+
+func (t *SessionTx) LoadPendingFileWrite(ctx context.Context) (bool, error) {
+	return t.q.EnvironmentFileWriteBlocksSession(ctx, t.session)
+}

services/core/internal/persistence/postgres/sessionpg/doc.go

@@ -1,10 +1,12 @@
-// Package sessionpg writes the Session changes that the sessions and items
-// packages decide to PostgreSQL. Its functions run inside the caller's Session
-// transaction, under the Session lock, on the transaction-bound queries: they
-// load the facts a decision reads, then apply the decision, allocating event
-// sequence positions, event IDs and Item positions, writing the public change
-// journal, Items, Turn usage and Artifacts, and pruning the journal. They decide
-// nothing.
+// Package sessionpg stores Sessions in PostgreSQL and decides nothing.
+// WithSession runs a Session transaction: it locks the Session, runs the
+// operation and prunes the journal. Inside a Session transaction, on its
+// queries, the participants load the facts the sessions and items decisions
+// read and apply what they decide: they lock the Session, allocate event
+// sequence positions, event IDs, Item positions and output indexes, and write
+// the public change journal, Items, Turn usage, Artifact settlement,
+// Environment state, input reservations and devices. SessionTx binds a Session
+// to its caller's transaction for the sessions procedures.
 package sessionpg
 
 import (