agentplane: give sampler Runner execution its own package

agentplane/central_sampler/BUILD.bazel

@@ -318,12 +318,6 @@ go_library(
     srcs = [
         "codemode.go",
         "codemode_host.go",
-        "exec_incarnation_fence.go",
-        "executor.go",
-        "executor_failure.go",
-        "executor_operations.go",
-        "executor_processes.go",
-        "executor_session.go",
         "protocol_adapter.go",
         "provider_recovery_runtime.go",
         "runtime.go",
@@ -335,8 +329,6 @@ go_library(
         "session_setup.go",
         "session_turn.go",
         "tool_invocation.go",
-        "unified_exec.go",
-        "view_image.go",
     ],
     importpath = "github.com/agentydragon/ducktape/agentplane/central_sampler",
     visibility = [
@@ -349,6 +341,7 @@ go_library(
         "//agentplane/central_sampler/internal/codemode",
         "//agentplane/central_sampler/internal/content",
         "//agentplane/central_sampler/internal/eventcodec",
+        "//agentplane/central_sampler/internal/executor",
         "//agentplane/central_sampler/internal/outputbuffer",
         "//agentplane/central_sampler/internal/promptimage",
         "//agentplane/central_sampler/internal/sessionstore",
@@ -392,8 +385,6 @@ go_test(
     srcs = [
         "anthropic_test.go",
         "codemode_test.go",
-        "executor_resume_test.go",
-        "executor_test.go",
         "integration_testmain_test.go",
         "notification_test.go",
         "provider_compaction_control_test.go",
@@ -403,7 +394,6 @@ go_test(
         "provider_turn_progress_test.go",
         "runtime_lifecycle_test.go",
         "settings_fixture_test.go",
-        "unified_exec_test.go",
     ],
     embed = [":central_sampler"],
     race = "on",
@@ -413,6 +403,7 @@ go_test(
         "//agentplane/central_sampler/historyexport",
         "//agentplane/central_sampler/internal/codemode",
         "//agentplane/central_sampler/internal/content",
+        "//agentplane/central_sampler/internal/executor",
         "//agentplane/central_sampler/internal/outputbuffer",
         "//agentplane/central_sampler/internal/sessionstore",
         "//agentplane/central_sampler/internal/settings",
@@ -449,7 +440,6 @@ go_test(
     name = "central_sampler_unit_test",
     size = "medium",
     srcs = [
-        "executor_evidence_test.go",
         "provider_authorization_test.go",
     ],
     embed = [":central_sampler"],

agentplane/central_sampler/README.md

@@ -21,6 +21,8 @@ connection to the Runner and does not create the logical Session.
 The shared child process, wire framing, request routing and logical-session lifecycle live in
 `internal/codemode`. The sampler binds each session to its lease-fenced journal and delegated tool
 handler; model tool adaptation, nested executor intents and result evidence remain sampler-owned.
+The concrete Runner adapter, including session handshake/rebind, process and file RPCs, cursor
+reconciliation and execution evidence, lives in `internal/executor`.
 
 The service, sampling loop and persistence are implemented in Go. History export lives in the
 `historyexport` package and reaches sampler state through its durable outbox interface. The
@@ -176,12 +178,15 @@ the extant bound Sandbox object; the standalone API can inspect/control its own
 ## Verification
 
 Run `//agentplane/central_sampler:central_sampler_unit_test` for the remaining runtime unit tests
-without starting PostgreSQL. `//agentplane/central_sampler:central_sampler_test` runs the runtime,
-RPC and executor integration cases; `//agentplane/central_sampler/internal/sessionstore:sessionstore_test`
-runs Store, recovery, outbox and migration-adoption cases. Leaf packages have colocated test targets;
-`bbr test //agentplane/central_sampler/...` includes them in the Go and automated acceptance sweep.
-`//agentplane/history_service/internal/ingress:integration_test` additionally checks real sampler
-export and History projection across recovery. The integration targets use real PostgreSQL,
+without starting PostgreSQL. `//agentplane/central_sampler:central_sampler_test` runs runtime and
+provider integration cases. `//agentplane/central_sampler/internal/executor:executor_test` covers
+Runner session/resume, cursor, image and no-replay behavior;
+`//agentplane/central_sampler/internal/sessionstore:sessionstore_test` covers Store, recovery,
+outbox and migration adoption. RPC auth/status and request-streaming tests live in
+`//agentplane/central_sampler/internal/rpc:unit_test` and `:integration_test`.
+Leaf packages have colocated test targets; `bbr test //agentplane/central_sampler/...` includes them
+in the Go and automated acceptance sweep. `//agentplane/history_service/internal/ingress:integration_test`
+additionally checks real sampler export and History projection across recovery. Integration targets use real PostgreSQL,
 pinned upstream host/executor processes, authenticated Gateway/History and scripted provider wire
 fixtures. Scripted acceptance does not establish live provider or subscription compatibility.
 Responses Lite compaction uses Codex's remote-v2 beta protocol. Anthropic uses native server-side

agentplane/central_sampler/codemode.go

@@ -8,6 +8,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/agentydragon/ducktape/agentplane/central_sampler/internal/executor"
 	"github.com/agentydragon/ducktape/agentplane/central_sampler/internal/sessionstore"
 
 	"github.com/agentydragon/ducktape/agentplane/central_sampler/internal/codemode"
@@ -25,15 +26,15 @@ type CodeMode struct {
 	*codemode.Session
 	store    *sessionstore.Store
 	owner    sessionstore.Owner
-	executor *Executor
+	executor *executor.Executor
 }
 
 func (r *CodeMode) Execute(ctx context.Context, call protocol.ToolInvocation) ([]protocol.ContentPart, error) {
 	prepared, err := r.prepareCall(ctx, call)
 	if err != nil {
 		// Codex's FunctionCallError::RespondToModel returns input errors to the model;
 		// they do not stop the turn or abandon unrelated live cells and processes.
-		var inputError *toolError
+		var inputError *toolcontract.Rejection
 		if errors.As(err, &inputError) {
 			return []protocol.ContentPart{{Type: "input_text", Text: "Error: " + inputError.Error()}}, nil
 		}
@@ -87,11 +88,11 @@ func (r *CodeMode) prepareCall(ctx context.Context, call protocol.ToolInvocation
 		return r.prepareWait(ctx, call.Input)
 	}
 	if call.Name != "exec" || call.Encoding != protocol.ExecEncodingFreeform {
-		return preparedCall{}, &toolError{fmt.Sprintf("unsupported model tool %q (%s)", call.Name, call.Encoding)}
+		return preparedCall{}, &toolcontract.Rejection{Message: fmt.Sprintf("unsupported model tool %q (%s)", call.Name, call.Encoding)}
 	}
 	source, err := toolcontract.ParseExecSource(call.Input)
 	if err != nil {
-		return preparedCall{}, &toolError{err.Error()}
+		return preparedCall{}, &toolcontract.Rejection{Message: err.Error()}
 	}
 	return r.prepareExecute(call.CallID, source)
 }
@@ -117,7 +118,7 @@ func (r *CodeMode) prepareExecute(callID string, source toolcontract.ExecSource)
 func (r *CodeMode) prepareJSONCall(ctx context.Context, call protocol.ToolInvocation) (preparedCall, error) {
 	parsed, err := toolcontract.ParseJSONExecInput(call.Input)
 	if err != nil {
-		return preparedCall{}, &toolError{err.Error()}
+		return preparedCall{}, &toolcontract.Rejection{Message: err.Error()}
 	}
 	if parsed.Wait != nil {
 		return r.prepareWaitArguments(ctx, *parsed.Wait)
@@ -128,7 +129,7 @@ func (r *CodeMode) prepareJSONCall(ctx context.Context, call protocol.ToolInvoca
 func (r *CodeMode) prepareWait(ctx context.Context, input string) (preparedCall, error) {
 	arguments, err := toolcontract.ParseWaitArguments(input)
 	if err != nil {
-		return preparedCall{}, &toolError{err.Error()}
+		return preparedCall{}, &toolcontract.Rejection{Message: err.Error()}
 	}
 	return r.prepareWaitArguments(ctx, arguments)
 }
@@ -147,11 +148,11 @@ func (r *CodeMode) prepareWaitArguments(ctx context.Context, arguments toolcontr
 		}
 	}
 	if !known {
-		return preparedCall{}, &toolError{"wait requires a known live cell_id; process session_id is not accepted"}
+		return preparedCall{}, &toolcontract.Rejection{Message: "wait requires a known live cell_id; process session_id is not accepted"}
 	}
 	nativeID, err := r.NativeCell(cellID)
 	if err != nil {
-		return preparedCall{}, &toolError{err.Error()}
+		return preparedCall{}, &toolcontract.Rejection{Message: err.Error()}
 	}
 	session := codemode.SessionOperation{Method: "session/wait", SessionID: r.ID()}
 	if arguments.Terminate {

agentplane/central_sampler/codemode_host.go

@@ -8,10 +8,12 @@ import (
 	"strings"
 	"time"
 
+	"github.com/agentydragon/ducktape/agentplane/central_sampler/internal/executor"
 	"github.com/agentydragon/ducktape/agentplane/central_sampler/internal/sessionstore"
 
 	"github.com/agentydragon/ducktape/agentplane/central_sampler/internal/codemode"
 	samplersettings "github.com/agentydragon/ducktape/agentplane/central_sampler/internal/settings"
+	"github.com/agentydragon/ducktape/agentplane/central_sampler/internal/toolcontract"
 	"github.com/agentydragon/ducktape/agentplane/central_sampler/protocol"
 	eventpb "github.com/agentydragon/ducktape/agentplane/proto/event"
 )
@@ -26,7 +28,7 @@ func NewCodeModeHost(parent context.Context, settings samplersettings.Settings,
 	return &CodeModeHost{Manager: codemode.NewManager(parent, settings.CodeModeHost, settings.Logs, time.Duration(settings.RPCTimeoutSeconds)*time.Second, settings.CodeModeObservationTimeout()), store: store}
 }
 
-func (m *CodeModeHost) OpenSession(ctx context.Context, conversationID string, owner sessionstore.Owner, executor *Executor) (*CodeMode, error) {
+func (m *CodeModeHost) OpenSession(ctx context.Context, conversationID string, owner sessionstore.Owner, executor *executor.Executor) (*CodeMode, error) {
 	if executor == nil {
 		return nil, errors.New("code-mode executor is required")
 	}
@@ -99,26 +101,26 @@ func (r *CodeMode) delegate(ctx context.Context, session *codemode.Session, requ
 		return empty, empty, err
 	}
 	result, err := r.executor.Invoke(ctx, invocation.ToolName.Name, invocation.Input)
-	var rejected *toolError
+	var rejected *toolcontract.Rejection
 	if errors.As(err, &rejected) {
 		// A known rejection settles the nested intent before rejecting the host promise.
-		var evidence any = map[string]any{"error": rejected.message}
-		var outcome *toolOutcomeError
+		var evidence any = map[string]any{"error": rejected.Message}
+		var outcome *executor.ToolOutcomeError
 		if errors.As(err, &outcome) {
-			evidence = outcome.evidence
+			evidence = outcome.Evidence()
 		}
 		if err := r.store.FinishNested(ctx, r.owner, intent, evidence); err != nil {
 			return empty, empty, err
 		}
-		reply := codemode.DelegateResult{Status: "error", Message: rejected.message}
+		reply := codemode.DelegateResult{Status: "error", Message: rejected.Message}
 		return reply, reply, nil
 	}
 	if err != nil {
 		return empty, empty, err
 	}
 	var evidence map[string]any
-	if viewed, ok := result.(*viewImageResult); ok {
-		evidence = viewed.evidence()
+	if viewed, ok := result.(*executor.ViewImageResult); ok {
+		evidence = viewed.Evidence()
 	} else if evidence, err = protocol.Object(result); err != nil {
 		return empty, empty, err
 	}

agentplane/central_sampler/codemode_test.go

@@ -16,10 +16,12 @@ import (
 	"testing"
 	"time"
 
+	"github.com/agentydragon/ducktape/agentplane/central_sampler/internal/executor"
 	"github.com/agentydragon/ducktape/agentplane/central_sampler/internal/sessionstore"
 
 	"github.com/agentydragon/ducktape/agentplane/central_sampler/internal/codemode"
 	samplersettings "github.com/agentydragon/ducktape/agentplane/central_sampler/internal/settings"
+	"github.com/agentydragon/ducktape/agentplane/central_sampler/internal/toolcontract"
 	"github.com/agentydragon/ducktape/agentplane/central_sampler/protocol"
 	"github.com/agentydragon/ducktape/agentplane/central_sampler/providers/responses"
 	eventpb "github.com/agentydragon/ducktape/agentplane/proto/event"
@@ -133,7 +135,7 @@ func TestCodeModeStoreFailureIsNotModelInputError(t *testing.T) {
 	output, err := runtime.Execute(t.Context(), protocol.ToolInvocation{
 		Encoding: protocol.ExecEncodingJSON, Name: "exec", Input: `{"cell_id":"fixture:cell"}`,
 	})
-	if err == nil || output != nil || errors.As(err, new(*toolError)) {
+	if err == nil || output != nil || errors.As(err, new(*toolcontract.Rejection)) {
 		t.Fatalf("store failure became model output: output=%+v error=%v", output, err)
 	}
 }
@@ -407,7 +409,7 @@ func TestCodeModeHostStampsOpenedHostWhenItFailsBeforeStamp(t *testing.T) {
 
 func openFakeSession(t *testing.T, manager *CodeModeHost, conversation string) *CodeMode {
 	t.Helper()
-	code, err := manager.OpenSession(context.Background(), conversation, sessionstore.Owner{ConversationID: conversation}, &Executor{})
+	code, err := manager.OpenSession(context.Background(), conversation, sessionstore.Owner{ConversationID: conversation}, &executor.Executor{})
 	if err != nil {
 		t.Fatalf("open logical session %q: %v", conversation, err)
 	}
@@ -423,7 +425,7 @@ func TestRecoveryUsesNewLogicalSessionWithoutRestartingSharedHost(t *testing.T)
 	}
 	manager, peers := newFakeCodeModeManager(t)
 	manager.store = store
-	first, err := manager.OpenSession(t.Context(), session.ConversationID, owner, &Executor{})
+	first, err := manager.OpenSession(t.Context(), session.ConversationID, owner, &executor.Executor{})
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -450,7 +452,7 @@ func TestRecoveryUsesNewLogicalSessionWithoutRestartingSharedHost(t *testing.T)
 	if err != nil {
 		t.Fatal(err)
 	}
-	second, err := manager.OpenSession(t.Context(), session.ConversationID, newOwner, &Executor{})
+	second, err := manager.OpenSession(t.Context(), session.ConversationID, newOwner, &executor.Executor{})
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -928,7 +930,7 @@ func newCodeModeAdvanceFixture(t *testing.T, call protocol.ToolInvocation) (*ses
 	}
 	manager, peers := newFakeCodeModeManager(t)
 	manager.store = store
-	code, err := manager.OpenSession(t.Context(), session.ConversationID, owner, &Executor{})
+	code, err := manager.OpenSession(t.Context(), session.ConversationID, owner, &executor.Executor{})
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -965,7 +967,7 @@ func newCodeModeAdvanceFixture(t *testing.T, call protocol.ToolInvocation) (*ses
 		RPCTimeoutSeconds: 2,
 	}
 	runtime := NewRuntime(store, settings, manager)
-	return store, owner, runtime, &liveHost{code: code, executor: &Executor{}}, code, peers()[0]
+	return store, owner, runtime, &liveHost{code: code, executor: &executor.Executor{}}, code, peers()[0]
 }
 
 func appendModelToolCall(t *testing.T, store *sessionstore.Store, owner sessionstore.Owner, call protocol.ToolInvocation) {
@@ -1305,7 +1307,7 @@ func TestCodeModeObservationBudgetsBoundWedgedHostWithoutExpiringCaller(t *testi
 			}
 			manager, peers := newFakeCodeModeManagerWithBudgets(t, 2*time.Second, 20*time.Millisecond)
 			manager.store = store
-			code, err := manager.OpenSession(t.Context(), session.ConversationID, owner, &Executor{})
+			code, err := manager.OpenSession(t.Context(), session.ConversationID, owner, &executor.Executor{})
 			if err != nil {
 				t.Fatal(err)
 			}

agentplane/central_sampler/integration_testmain_test.go

@@ -175,6 +175,31 @@ func openSessionForStore(t *testing.T, store *sessionstore.Store, setup string)
 	return session
 }
 
+func claimedExecutorStore(t *testing.T) (*sessionstore.Store, sessionstore.Owner) {
+	t.Helper()
+	store := newTestStore(t)
+	session := openSessionForStore(t, store, "")
+	owner, err := store.Claim(t.Context(), session.ConversationID, time.Minute)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := store.SetExecutorSession(t.Context(), owner, "retained-executor", nil); err != nil {
+		t.Fatal(err)
+	}
+	return store, owner
+}
+
+func assertNoUnresolvedNestedIntents(t *testing.T, store *sessionstore.Store, owner sessionstore.Owner) {
+	t.Helper()
+	var count int
+	if err := store.DB().QueryRowContext(t.Context(), "SELECT count(*) FROM sampler_intent WHERE conversation_id=$1 AND result IS NULL", owner.ConversationID).Scan(&count); err != nil {
+		t.Fatal(err)
+	}
+	if count != 0 {
+		t.Fatalf("unresolved nested intents=%d", count)
+	}
+}
+
 func admitInput(t *testing.T, store *sessionstore.Store, sessionID, commandID, text string) {
 	t.Helper()
 	if _, err := store.Admit(t.Context(), sessionID, &commandpb.Command{CommandId: commandID, Operation: &commandpb.Command_SubmitInput{SubmitInput: &commandpb.SubmitInput{Text: text}}}, "fixture-account", nil); err != nil {

agentplane/central_sampler/internal/executor/BUILD.bazel

@@ -0,0 +1,69 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "executor",
+    srcs = [
+        "executor.go",
+        "failure.go",
+        "incarnation_fence.go",
+        "operations.go",
+        "processes.go",
+        "session.go",
+        "unified_exec.go",
+        "view_image.go",
+    ],
+    importpath = "github.com/agentydragon/ducktape/agentplane/central_sampler/internal/executor",
+    visibility = ["//agentplane/central_sampler:__subpackages__"],
+    deps = [
+        "//agentplane/central_sampler/api:sampler_go_proto",
+        "//agentplane/central_sampler/internal/outputbuffer",
+        "//agentplane/central_sampler/internal/sessionstore",
+        "//agentplane/central_sampler/internal/settings",
+        "//agentplane/central_sampler/internal/toolcontract",
+        "//agentplane/central_sampler/internal/tooloutput",
+        "//agentplane/central_sampler/protocol",
+        "//agentplane/env/runtime/execution:executor_go_proto",
+        "//agentplane/grpc/bearer",
+        "//agentplane/proto:event_go_proto",
+        "@com_github_google_uuid//:uuid",
+        "@org_golang_google_grpc//:grpc",
+        "@org_golang_google_grpc//codes",
+        "@org_golang_google_grpc//credentials/insecure",
+        "@org_golang_google_grpc//status",
+        "@org_golang_google_protobuf//proto",
+    ],
+)
+
+go_test(
+    name = "executor_test",
+    size = "medium",
+    srcs = [
+        "executor_evidence_test.go",
+        "executor_resume_test.go",
+        "executor_test.go",
+        "integration_testmain_test.go",
+        "unified_exec_test.go",
+    ],
+    embed = [":executor"],
+    race = "on",
+    tags = ["requires_docker"],
+    deps = [
+        "//agentplane/central_sampler/api:sampler_go_proto",
+        "//agentplane/central_sampler/internal/outputbuffer",
+        "//agentplane/central_sampler/internal/sessionstore",
+        "//agentplane/central_sampler/internal/toolcontract",
+        "//agentplane/central_sampler/migrations",
+        "//agentplane/central_sampler/protocol",
+        "//agentplane/env/runtime/execution:executor_go_proto",
+        "//agentplane/proto:event_go_proto",
+        "//agentplane/proto:identity_go_proto",
+        "//agentplane/proto:session_go_proto",
+        "//agentplane/testing/postgres",
+        "@com_github_google_uuid//:uuid",
+        "@com_github_lib_pq//:pq",
+        "@org_golang_google_grpc//:grpc",
+        "@org_golang_google_grpc//codes",
+        "@org_golang_google_grpc//status",
+        "@org_golang_google_protobuf//proto",
+    ],
+)