Move Item, Subagent and Artifact reads and Artifact staging into sessions and sessionpg

services/core/IMPLEMENTATION.md

@@ -34,6 +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 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`.
+- `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
 
@@ -49,15 +50,15 @@ On the Beta group, the OpenAI-Beta check (exactly one `agents=v1` value) runs be
 
 - Stored strings other than metadata rely on PostgreSQL rejecting U+0000 and invalid UTF-8: `pgunit.IsUnstorableText` detects SQLSTATE `22021` (text parameter) and `22P05` (`\u0000` in jsonb), and the adapter returns `textvalue.ErrUnstorable`, the 400 unstorable-text error, including for query filters such as `agent_id`. The failing statement aborts its transaction, so keep each request's writes in one transaction.
 - An `after` cursor that cannot name a resource on a lookup list (Agents, Sessions, Turns, Templates, Vaults, Credentials) resolves through `pgunit.LookupCursor` to the never-assigned maximum UUID and runs the normal lookup, so storage failures and missing rows behave as for a well-formed cursor. Resolve every cursor only inside its already resolved parent and tenant.
-- Lists whose parent and cursor lookups are separate statements (Artifacts, and Skill versions in `skills`) re-check the parent before reporting a cursor 400, so a parent deleted in between still returns its 404. Item and Subagent lists read both inside one locked Session transaction. `skillpg` looks up a Skill version cursor tenant-wide so `skills` can tell another Skill's version from a missing one; another tenant's version stays missing.
+- Lists whose parent and cursor lookups are separate statements (Artifacts in `sessionpg`, and Skill versions in `skills`) re-check the parent before reporting a cursor 400, so a parent deleted in between still returns its 404. `sessionpg` reads Item and Subagent lists and their cursors inside one locked Session transaction. `skillpg` looks up a Skill version cursor tenant-wide so `skills` can tell another Skill's version from a missing one; another tenant's version stays missing.
 
 ## Source Files and Artifacts
 
 Source Files are Project resources with a lifecycle independent of copied workspace files. `files` owns their vocabulary, the upload envelope and 512 MiB content bound, the list rules and the create and delete use cases; `filepg` stores immutable source metadata and PostgreSQL large objects in Core's database with the pinned pgx driver. Upload validation, metadata insertion, the bytes and the write audit commit atomically; deletion removes the metadata, unlinks the object and records the audit in one transaction. Keep OIDs private and authorize every metadata, content and delete lookup by tenant before opening a body. Write content through `pgunit`'s large-object writer, which streams bounded chunks and reports the size and SHA-256 digest; never hold an entire upload in memory or use a filename as a filesystem path. A direct download of the `user_data` purpose is rejected after the tenant-scoped metadata lookup, while workspace copies keep their authorized `filepg` read and Session initialization copies read the object inside Session creation's store transaction. A read-only repeatable-read transaction preserves an admitted source across concurrent deletion; resolve that snapshot before entering the Environment write path, and a later deletion never undoes a completed workspace copy. Bound request and transaction lifetimes, roll back incomplete bodies and never retry an ambiguous commit automatically. Backups must include PostgreSQL large objects, and a schema rollback must not orphan them.
 
 Session Artifacts are immutable published copies, separate from live workspace files and source Files. The private output exporter reuses the authorized workspace path boundary and streams bounded bytes; publication requires complete capture and confirmed helper and transport success, not merely valid archive syntax. The daemon owns and drains the exporter's stdout pipe separately from child reaping, so pull-transport backpressure cannot consume the process-exit I/O deadline; after helper exit, each pipe read has one second, reset after consumer delays, which rejects inherited pipes that never close. Cancellation closes the owned reader and the dispatch consumer, then waits for the child. Never extract an output archive into Core's filesystem or hold the execution lease through a large transfer.
 
-Capture bytes into private large objects without a Session admission lock. Before capture, seal native input under that lock with the private capture marker; later messages reuse the Environment input reservation and wait for the next Turn, and the public Turn stays in progress until publication settles. Directory reads during capture use an independent read-only preparation, not the released native Run. After a confirmed export, lock and recheck the live Turn, then publish the metadata in the same transaction as the Turn's completion. `sessions.EndTurn` decides Artifact publication and `sessionpg` applies it in the Turn's terminal transaction, not the capture transaction: capture commits and releases the Session lock before the Turn completes, and the terminal transaction holds the Session lock that also orders Artifact deletion. Drop staged rows whose SHA-256 equals the newest remaining published Artifact for the path, ordered by the producing Turn's creation time and then ID (publication time can come from the Runtime and does not order Turns), and unlink their large objects in the same transaction. Published rows are never modified. Failed and cancelled Turns discard private objects, and Session deletion removes private and published copies. The exporter skips output symlinks by their `lstat` type without following them; hard links, other special files, device crossings and concurrent changes reject the capture. Stored reads are authorized independently of Environment availability, so published Artifacts outlive the Environment.
+Capture bytes into private large objects without a Session admission lock. Before capture, seal native input under that lock with the private capture marker; later messages reuse the Environment input reservation and wait for the next Turn, and the public Turn stays in progress until publication settles. Directory reads during capture use an independent read-only preparation, not the released native Run. `sessions.StageTurnArtifacts` stages a confirmed export as a pooled use case, never on the execution lease: it authorizes the Environment before reading the export, admits only distinct regular files under `outputs/` within the capture limits followed by nothing but zero padding, and after the transfer takes the Session lock through `sessionpg.LockSession` and rechecks that the Turn is in progress without a cancellation request. `sessionpg` streams each file into a large object in that one transaction, so a rejected or failed export rolls back its content. The metadata is published in the same transaction as the Turn's completion. `sessions.EndTurn` decides Artifact publication and `sessionpg` applies it in the Turn's terminal transaction, not the capture transaction: capture commits and releases the Session lock before the Turn completes, and the terminal transaction holds the Session lock that also orders Artifact deletion. Drop staged rows whose SHA-256 equals the newest remaining published Artifact for the path, ordered by the producing Turn's creation time and then ID (publication time can come from the Runtime and does not order Turns), and unlink their large objects in the same transaction. Published rows are never modified. Failed and cancelled Turns discard private objects, and Session deletion removes private and published copies. The exporter skips output symlinks by their `lstat` type without following them; hard links, other special files, device crossings and concurrent changes reject the capture. `sessionpg` serves stored reads authorized independently of Environment availability, so published Artifacts outlive the Environment, and reads content from one snapshot, so a read admitted before a deletion finishes. Deletion takes the Session lock, unlinks the content and records the write audit in one transaction.
 
 - A malformed `environment_id` Artifact filter resolves to the never-assigned maximum UUID, so it matches nothing without a text comparison.
 

services/core/cmd/server/http_routes_test.go

@@ -102,6 +102,7 @@ func daemonComposition(t testing.TB) http.Handler {
 		Items:           struct{ api.Items }{},
 		Subagents:       struct{ api.Subagents }{},
 		Artifacts:       struct{ api.Artifacts }{},
+		ArtifactsReader: struct{ api.ArtifactsReader }{},
 		SessionAdmin:    struct{ api.SessionAdmin }{}, Environments: struct{ api.Environments }{}, 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 }{},

services/core/cmd/server/main.go

@@ -48,6 +48,7 @@ import (
 	"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/persistence/postgres/projectpg"
+	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/persistence/postgres/sessionpg"
 	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/persistence/postgres/skillpg"
 	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/persistence/postgres/templatepg"
 	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/persistence/postgres/vaultpg"
@@ -60,6 +61,7 @@ import (
 	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/runtimeobs"
 	observationstoreresolver "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/runtimeobs/storeresolver"
 	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sandbox/providers"
+	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sessions"
 	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/skills"
 	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/store"
 	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/vaults"
@@ -162,6 +164,11 @@ func run() error {
 	if err != nil {
 		return err
 	}
+	sessionStore := sessionpg.New(units)
+	sessionService, err := sessions.NewService(sessionStore)
+	if err != nil {
+		return err
+	}
 	installation, err := installationFacts(public)
 	if err != nil {
 		return err
@@ -283,6 +290,8 @@ func run() error {
 	if registry != nil {
 		dispatcher := &execution.Dispatcher{Store: executionStore, Registry: registry,
 			Credentials: vaultService, Observer: modelConfigurationStore, Deployment: deploymentService,
+			Sessions:        sessionService,
+			SessionsReader:  sessionStore,
 			ManagedRuntimes: managed, MaxConcurrentExecutions: concurrency}
 		lease, err := pgunit.AcquireLease(ctx, pool)
 		if err != nil {
@@ -373,9 +382,10 @@ func run() error {
 		SessionCreation: executionStore,
 		SessionEvents:   executionStore,
 		Turns:           executionStore,
-		Items:           executionStore,
-		Subagents:       executionStore,
-		Artifacts:       executionStore,
+		Items:           sessionStore,
+		Subagents:       sessionStore,
+		Artifacts:       sessionService,
+		ArtifactsReader: sessionStore,
 		SessionAdmin:    executionStore,
 		Environments:    executionStore, ExecutorConnections: executorConnections{store: executionStore, registry: registry},
 		Admin: executionStore, AdminAudit: auditStore, WriteAudit: auditStore, Metrics: metrics,

services/core/internal/api/dependencies.go

@@ -49,6 +49,7 @@ type Dependencies struct {
 	Items                Items
 	Subagents            Subagents
 	Artifacts            Artifacts
+	ArtifactsReader      ArtifactsReader
 	SessionAdmin         SessionAdmin
 	Environments         Environments
 	ExecutorConnections  ExecutorConnections
@@ -136,7 +137,9 @@ func (d Dependencies) validate() error {
 		field{"Turns", d.Turns},
 		field{"Items", d.Items},
 		field{"Subagents", d.Subagents},
-		field{"Artifacts", d.Artifacts}, field{"SessionAdmin", d.SessionAdmin}, field{"Environments", d.Environments},
+		field{"Artifacts", d.Artifacts},
+		field{"ArtifactsReader", d.ArtifactsReader},
+		field{"SessionAdmin", d.SessionAdmin}, field{"Environments", d.Environments},
 		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

@@ -35,6 +35,7 @@ type testFakes struct {
 	items                  *fakeItems
 	subagents              *fakeSubagents
 	artifacts              *fakeArtifacts
+	artifactsReader        *fakeArtifactsReader
 	sessionAdmin           *fakeSessionAdmin
 	environments           *fakeEnvironments
 	executorConnections    *fakeExecutorConnections
@@ -78,8 +79,10 @@ func testDependencies(t testing.TB) (Dependencies, *testFakes) {
 		sessionEvents:   &fakeSessionEvents{t: t},
 		turns:           &fakeTurns{t: t},
 		items:           &fakeItems{t: t},
-		subagents:       &fakeSubagents{t: t}, artifacts: &fakeArtifacts{t: t},
-		sessionAdmin: &fakeSessionAdmin{t: t}, environments: &fakeEnvironments{t: t}, executorConnections: &fakeExecutorConnections{t: t},
+		subagents:       &fakeSubagents{t: t},
+		artifacts:       &fakeArtifacts{t: t},
+		artifactsReader: &fakeArtifactsReader{t: t},
+		sessionAdmin:    &fakeSessionAdmin{t: t}, environments: &fakeEnvironments{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},
@@ -103,8 +106,11 @@ func testDependencies(t testing.TB) (Dependencies, *testFakes) {
 		SessionEvents:   f.sessionEvents,
 		Turns:           f.turns,
 		Items:           f.items,
-		Subagents:       f.subagents, Artifacts: f.artifacts, SessionAdmin: f.sessionAdmin,
-		Environments: f.environments, ExecutorConnections: f.executorConnections, Admin: f.admin, AdminAudit: f.adminAudit, WriteAudit: f.writeAudit,
+		Subagents:       f.subagents,
+		Artifacts:       f.artifacts,
+		ArtifactsReader: f.artifactsReader,
+		SessionAdmin:    f.sessionAdmin,
+		Environments:    f.environments, 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/fakes_test.go

@@ -131,40 +131,44 @@ func (f *fakeAgentsReader) GetAgentWithModelProvider(a0 context.Context, a1 stri
 
 type fakeArtifacts struct {
 	t                     testing.TB
-	getSessionArtifact    func(context.Context, string, string, string) (sessions.Artifact, error)
-	listSessionArtifacts  func(context.Context, string, string, string, string, int, bool) (sessions.ArtifactPage, error)
-	readSessionArtifact   func(context.Context, string, string, string, func(sessions.Artifact, io.Reader) error) error
-	deleteSessionArtifact func(context.Context, string, string, string) error
+	deleteSessionArtifact func(context.Context, sessions.DeleteSessionArtifactCommand) error
 }
 
-func (f *fakeArtifacts) GetSessionArtifact(a0 context.Context, a1 string, a2 string, a3 string) (sessions.Artifact, error) {
+func (f *fakeArtifacts) DeleteSessionArtifact(a0 context.Context, a1 sessions.DeleteSessionArtifactCommand) error {
+	if f.deleteSessionArtifact == nil {
+		unexpectedCall(f.t, "DeleteSessionArtifact")
+	}
+	return f.deleteSessionArtifact(a0, a1)
+}
+
+type fakeArtifactsReader struct {
+	t                    testing.TB
+	getSessionArtifact   func(context.Context, string, string, string) (sessions.Artifact, error)
+	listSessionArtifacts func(context.Context, string, string, string, string, int, bool) (sessions.ArtifactPage, error)
+	readSessionArtifact  func(context.Context, string, string, string, func(sessions.Artifact, io.Reader) error) error
+}
+
+func (f *fakeArtifactsReader) GetSessionArtifact(a0 context.Context, a1 string, a2 string, a3 string) (sessions.Artifact, error) {
 	if f.getSessionArtifact == nil {
 		unexpectedCall(f.t, "GetSessionArtifact")
 	}
 	return f.getSessionArtifact(a0, a1, a2, a3)
 }
 
-func (f *fakeArtifacts) ListSessionArtifacts(a0 context.Context, a1 string, a2 string, a3 string, a4 string, a5 int, a6 bool) (sessions.ArtifactPage, error) {
+func (f *fakeArtifactsReader) ListSessionArtifacts(a0 context.Context, a1 string, a2 string, a3 string, a4 string, a5 int, a6 bool) (sessions.ArtifactPage, error) {
 	if f.listSessionArtifacts == nil {
 		unexpectedCall(f.t, "ListSessionArtifacts")
 	}
 	return f.listSessionArtifacts(a0, a1, a2, a3, a4, a5, a6)
 }
 
-func (f *fakeArtifacts) ReadSessionArtifact(a0 context.Context, a1 string, a2 string, a3 string, a4 func(sessions.Artifact, io.Reader) error) error {
+func (f *fakeArtifactsReader) ReadSessionArtifact(a0 context.Context, a1 string, a2 string, a3 string, a4 func(sessions.Artifact, io.Reader) error) error {
 	if f.readSessionArtifact == nil {
 		unexpectedCall(f.t, "ReadSessionArtifact")
 	}
 	return f.readSessionArtifact(a0, a1, a2, a3, a4)
 }
 
-func (f *fakeArtifacts) DeleteSessionArtifact(a0 context.Context, a1 string, a2 string, a3 string) error {
-	if f.deleteSessionArtifact == nil {
-		unexpectedCall(f.t, "DeleteSessionArtifact")
-	}
-	return f.deleteSessionArtifact(a0, a1, a2, a3)
-}
-
 type fakeConfigurationDiscovery struct {
 	t                     testing.TB
 	discoverConfiguration func(ctx context.Context, provider string, input sandbox.ConfigurationDiscoveryInput) (json.RawMessage, error)