Move function calls and results into sessions and sessionpg

services/core/IMPLEMENTATION.md

@@ -16,13 +16,13 @@ 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. 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.
+`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. `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`. `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`.
 
-`cmd/server` owns the execution lease. It acquires one `pgunit.Lease`, builds every lease-bound adapter on it, and passes the lease and those adapters together as one `execution.Owner` to `execution.StartWorker`. If anything fails before that call, `cmd/server` closes the lease. From that call the Worker owns cleanup: a failed start closes the lease before it returns, and a started Worker closes it after `Run` has cancelled and drained its work. Each close runs under its own bounded deadline, independent of the cancelled request or run. Lease-bound adapters and the store writer borrow the lease and never close it, and the Worker uses the lease only through `Owner.Lease`, never through an adapter. Store integration tests start the Worker the same way through `startWorker`.
+`cmd/server` owns the execution lease. It acquires one `pgunit.Lease`, builds every lease-bound adapter on it, and passes the lease and those adapters together as one `execution.Owner` to `execution.StartWorker`, which binds the Dispatcher to the Owner's execution operations through `Dispatcher.Bind`. If anything fails before that call, `cmd/server` closes the lease. From that call the Worker owns cleanup: a failed start closes the lease before it returns, and a started Worker closes it after `Run` has cancelled and drained its work. Each close runs under its own bounded deadline, independent of the cancelled request or run. Lease-bound adapters and the store writer borrow the lease and never close it, and the Worker uses the lease only through `Owner.Lease`, never through an adapter. Store integration tests start the Worker the same way through `startWorker`, and bind a Dispatcher that runs Turns without a Worker through `Dispatcher.Bind`.
 
 Domain owners, each with its PostgreSQL adapter under `internal/persistence/postgres`:
 
@@ -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.
+- `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 run through `sessions.ExecutionOperations` on `sessionpg.NewExecution(lease)`, which the Worker receives as `execution.Owner.Sessions`.
 
 ## Request handling
 
@@ -157,7 +157,7 @@ Core replaces a Turn's usage with each complete valid token breakdown, which `se
 
 Subagent observations use the common types in `internal/agentdaemon/proto/subagents.go`. Core assigns public IDs and projects them under the Session lock and the leased execution journal; native names, history parsing and outcome proof stay in the adapters. Child Turns have their own table, separate from Core's queue, and Session Turn reads and the Session event stream carry root work only. The migration-defined `public_execution_turns` view has no public reader; do not reintroduce mixed Session Turn pages. Repeated effects are idempotent, root output freezes first, and child Items are delivered before their terminal Turn snapshot.
 
-Function calls are stored per tenant, Session and Turn with immutable public and executor call identities and arguments. Result admission and application receipts serialize on the Session lock with cancellation and terminal transitions. The store keeps the complete caller-validated result, including omitted versus null error and output and ordered text and image parts; identical retries return the saved decision and changed results conflict. Recording a call moves the Turn to `waiting`, and the last application receipt resumes it. A function result may join a message or cancel batch: its explicit Turn and call identity select an existing call, admission never creates a Turn for it, and the result and its input retry record commit in one transaction. Targets resolve only after the tenant Session lookup; an unknown call or a call of another Turn is 400 `invalid_request_error`. The input cursor skips function results, whose own receipts decide application. A Session with functions waits for a device that declares `function_tools`, and Core delivers each saved result once per live dispatch, appending a non-null error as a final text part because the native result has no error field.
+Function calls are stored per tenant, Session and Turn with immutable public and executor call identities and arguments. Result admission and application receipts serialize on the Session lock with cancellation and terminal transitions. `sessions.AdmitFunctionResult` keeps the complete caller-validated result, including omitted versus null error and output and ordered text and image parts; identical retries return the saved decision and changed results conflict. Recording a call through `sessions.ExecutionOperations` moves the Turn to `waiting`, and the last application receipt resumes it. A function result may join a message or cancel batch: its explicit Turn and call identity select an existing call, admission never creates a Turn for it, and the result and its input retry record commit in one transaction. Targets resolve only after the tenant Session lookup; an unknown call or a call of another Turn is 400 `invalid_request_error`. The input cursor skips function results, whose own receipts decide application. A Session with functions waits for a device that declares `function_tools`, and Core delivers each saved result once per live dispatch, appending a non-null error as a final text part because the native result has no error field.
 
 Neutral tool and message observations go into the journal before Core projects public Items; the Item projector validates the shared observation contract and never decodes engine-native snapshots. `command_output` fragments update Items only for a command already indexed in the same Turn, and commit with their `agent.output.command_execution_output.delta` events. Completion output replaces accumulated drafts, terminal Items ignore late fragments, and cancellation keeps partial output without inventing a successful completion.
 

services/core/cmd/server/main.go

@@ -301,11 +301,16 @@ func run() error {
 		if err != nil {
 			return errors.Join(err, lease.Close(ctx))
 		}
+		sessionExecution, err := sessions.NewExecutionOperations(sessionpg.NewExecution(lease))
+		if err != nil {
+			return errors.Join(err, lease.Close(ctx))
+		}
 		// From this call on the Worker closes the lease, even when it fails to start.
 		worker, err = execution.StartWorker(ctx, dispatcher, execution.Owner{
 			Lease:      lease,
 			Store:      store.NewExecution(executionStore, lease),
 			Deployment: deploymentExecution,
+			Sessions:   sessionExecution,
 		})
 		if err != nil {
 			return err

services/core/internal/execution/delivery.go

@@ -105,7 +105,7 @@ func (d *Dispatcher) deliver(ctx context.Context, tenantID, sessionID string, pe
 	var pending *pendingInput
 	var cancelSent time.Time
 	var cancelReply <-chan cancellationResult
-	functions := &functionExchange{kind: request.AgentKind, store: d.Store, tenant: tenantID, session: sessionID, turn: request.RunID, tools: request.FunctionTools}
+	functions := &functionExchange{kind: request.AgentKind, store: d.Store, sessions: d.sessionExecution, tenant: tenantID, session: sessionID, turn: request.RunID, tools: request.FunctionTools}
 	done := false
 	cancelCtx, stopCancellation := context.WithCancel(ctx)
 	defer stopCancellation()

services/core/internal/execution/deployment_provider_observations_test.go

@@ -73,7 +73,11 @@ func newFinishObservationFixture(t *testing.T, maxConnections int32) finishObser
 	if err != nil {
 		t.Fatal(err)
 	}
-	return finishObservationFixture{s, owner.Store, owner.Lease, pool, defaults, tenant, session, Dispatcher{Store: owner.Store, Observer: defaults}}
+	dispatcher, err := (&Dispatcher{Observer: defaults}).Bind(owner)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return finishObservationFixture{s, owner.Store, owner.Lease, pool, defaults, tenant, session, *dispatcher}
 }
 func (f finishObservationFixture) start(t *testing.T) sessions.InputReceipt {
 	t.Helper()

services/core/internal/execution/dispatcher.go

@@ -29,6 +29,9 @@ type Snapshot struct {
 
 type Dispatcher struct {
 	notifications *executionNotifications
+	// sessionExecution runs the Session execution operations on the lease;
+	// Bind sets it from Owner.Sessions.
+	sessionExecution *sessions.ExecutionOperations
 	Policy
 	Store    *store.Store
 	Registry *runtimegateway.Registry

services/core/internal/execution/functions.go

@@ -51,6 +51,7 @@ type functionReply struct {
 
 type functionExchange struct {
 	store                 *store.Store
+	sessions              *sessions.ExecutionOperations
 	tenant, session, turn string
 	kind                  string
 	tools                 []proto.FunctionTool
@@ -70,7 +71,7 @@ func (f *functionExchange) record(ctx context.Context, env proto.Envelope) error
 	if !declared {
 		return errors.New("undeclared function callback")
 	}
-	err := f.store.RecordFunctionCall(ctx, f.tenant, f.session, f.turn, sessions.FunctionCall{
+	err := f.sessions.RecordFunctionCall(ctx, f.tenant, f.session, f.turn, sessions.FunctionCall{
 		CallID: items.Identity(f.turn, "tool:"+call.CallID), ExecutorCallID: call.CallID, Name: call.Name, Arguments: call.Arguments,
 	})
 	return f.unlessCancelling(ctx, err)
@@ -80,7 +81,7 @@ func (f *functionExchange) start(ctx context.Context, peer *runtimegateway.Sessi
 	if f.reply != nil || len(f.tools) == 0 {
 		return nil
 	}
-	calls, err := f.store.PendingFunctionCalls(ctx, f.tenant, f.session, f.turn)
+	calls, err := f.sessions.PendingFunctionCalls(ctx, f.tenant, f.session, f.turn)
 	if err != nil {
 		return err
 	}
@@ -119,7 +120,7 @@ func (f *functionExchange) confirm(ctx context.Context, reply functionReply) err
 	if !reply.ack.Applied {
 		return fmt.Errorf("function result not applied: %s", reply.ack.ErrorCode)
 	}
-	return f.unlessCancelling(ctx, f.store.ConfirmFunctionResult(ctx, f.tenant, f.session, f.turn, id))
+	return f.unlessCancelling(ctx, f.sessions.ConfirmFunctionResult(ctx, f.tenant, f.session, f.turn, id))
 }
 
 func (f *functionExchange) unlessCancelling(ctx context.Context, err error) error {
@@ -136,7 +137,7 @@ func (f *functionExchange) complete(ctx context.Context) error {
 	if len(f.tools) == 0 {
 		return nil
 	}
-	calls, err := f.store.PendingFunctionCalls(ctx, f.tenant, f.session, f.turn)
+	calls, err := f.sessions.PendingFunctionCalls(ctx, f.tenant, f.session, f.turn)
 	if err != nil {
 		return err
 	}