feat(plugins): add native-worker capabilities for remote executor plugins

apps/backend/cmd/agentctl/main.go

@@ -284,7 +284,7 @@ func run(cfg *config.Config, log *logger.Logger) {
 		}
 		// Shutdown all instances first so any final ACP frames they emit during
 		// teardown are still captured by the (still-live) debug writers.
-		if err := instMgr.Shutdown(ctx); err != nil {
+		if err := controlServer.ShutdownInstances(ctx); err != nil {
 			log.Error("error shutting down instance manager", zap.Error(err))
 		}
 		// Then flush + close all debug-log writers as the final logging step.

apps/backend/cmd/plugin-fixture/executor_provider.go

@@ -199,6 +199,12 @@ func (p *fixturePlugin) DestroyExecutorEnvironment(ctx context.Context, req *plu
 	if err := ctx.Err(); err != nil {
 		return nil, err
 	}
+	if refusal := fixtureDrainRefusal(req); refusal != nil {
+		return &pluginsdk.DestroyExecutorEnvironmentResponse{Error: refusal}, nil
+	}
+	if refusal := p.checkpointFixtureCleanup(ctx, req); refusal != nil {
+		return &pluginsdk.DestroyExecutorEnvironmentResponse{Error: refusal}, nil
+	}
 	if consumeFixtureFailureBarrier(p.dataDir, req.GetContext().GetOperationId(), fixtureBarrierUnconfirmedCleanupOnce) {
 		return &pluginsdk.DestroyExecutorEnvironmentResponse{Error: fixtureProviderError("cleanup_unconfirmed", "provider.fixture.cleanupUnconfirmed")}, nil
 	}
@@ -232,7 +238,7 @@ func (p *fixturePlugin) allocateFixtureExecutor(request *pluginsdk.ExecutorProvi
 	}
 	if existing, ok := p.executorState.Environments[request.GetOperationId()]; ok {
 		if existing.EnvironmentID != request.GetEnvironmentId() || existing.InputDigest != request.GetInputDigest() {
-			return fixtureExecutorResource{}, errors.New("plugin-fixture: operation identity changed after allocation")
+			return fixtureExecutorResource{}, pluginsdk.CategorizeActionError(pluginsdk.ActionErrorConflict, errors.New("plugin-fixture: operation identity changed after allocation"))
 		}
 		return existing, nil
 	}
@@ -558,3 +564,80 @@ func consumeFixtureFailureBarrier(dataDir, operationID, barrier string) bool {
 	_ = file.Close()
 	return true
 }
+
+// checkpointFixtureCleanup records cleanup_progress through the host before the
+// destroy mutates anything. A destroy under a cleanup claim whose checkpoint the
+// host refuses fails closed, as a real provider's does.
+func (p *fixturePlugin) checkpointFixtureCleanup(ctx context.Context, req *pluginsdk.DestroyExecutorEnvironmentRequest) *pluginsdk.ExecutorProviderError {
+	if req.GetCleanupClaim() == "" || !fixtureDeclaresCapability(fixtureCapabilityDestroyCheckpoints) {
+		return nil
+	}
+	host, ok := p.Host().(pluginsdk.ExecutorProviderHost)
+	if !ok {
+		return fixtureProviderError("cleanup_checkpoint_refused", "provider.fixture.cleanupCheckpointRefused")
+	}
+	if _, err := host.ExecutorProvider().CheckpointExecutorResource(ctx, &pluginsdk.CheckpointExecutorResourceRequest{
+		Context: req.GetContext(), Resource: req.GetResource(), Phase: "cleanup_progress", Sequence: 1,
+		Claim: req.GetCleanupClaim(),
+	}); err != nil {
+		return fixtureProviderError("cleanup_checkpoint_refused", "provider.fixture.cleanupCheckpointRefused")
+	}
+	return nil
+}
+
+const (
+	fixtureCapabilitiesEnv               = "KANDEV_PLUGIN_FIXTURE_CAPABILITIES"
+	fixtureCapabilityDestroyCheckpoints  = "destroy_checkpoints"
+	fixtureCapabilityDrainReceipt        = "drain_receipt"
+	fixtureLaunchFailedCleanupReason     = "launch_failed"
+	fixtureDrainOutcomeDrained           = "drained"
+	fixtureDrainOutcomeNotAttempted      = "not_attempted"
+	fixtureDefaultDeclaredCapabilityList = fixtureCapabilityDestroyCheckpoints + "," + fixtureCapabilityDrainReceipt
+)
+
+// fixtureDeclaresCapability reports whether the fixture acts as a provider that
+// requires the named host capability. The set is configuration, so a test can
+// run the fixture as a provider that declares none.
+func fixtureDeclaresCapability(name string) bool {
+	list, ok := os.LookupEnv(fixtureCapabilitiesEnv)
+	if !ok {
+		list = fixtureDefaultDeclaredCapabilityList
+	}
+	for _, declared := range strings.Split(list, ",") {
+		if strings.TrimSpace(declared) == name {
+			return true
+		}
+	}
+	return false
+}
+
+// fixtureDrainRefusal applies the destroy-time drain rules of a provider that
+// requires drain_receipt: not_attempted proves "no runtime" only under the
+// launch cleanup reason, and a drained outcome carries a receipt bound to the
+// cleanup claim of this destroy.
+func fixtureDrainRefusal(req *pluginsdk.DestroyExecutorEnvironmentRequest) *pluginsdk.ExecutorProviderError {
+	if !fixtureDeclaresCapability(fixtureCapabilityDrainReceipt) {
+		return nil
+	}
+	refuse := func() *pluginsdk.ExecutorProviderError {
+		return fixtureProviderError("drain_evidence_refused", "provider.fixture.drainEvidenceRefused")
+	}
+	switch req.GetDrainOutcome() {
+	case fixtureDrainOutcomeDrained:
+		receipt := req.GetDrainReceipt()
+		if receipt == nil || receipt.GetReceiptId() == "" || receipt.GetCleanupClaim() != req.GetCleanupClaim() {
+			return refuse()
+		}
+	case fixtureDrainOutcomeNotAttempted:
+		if req.GetCleanupReason() != fixtureLaunchFailedCleanupReason || req.GetDrainReceipt() != nil {
+			return refuse()
+		}
+	case "unreachable", "uncertain":
+		if req.GetDrainReceipt() != nil {
+			return refuse()
+		}
+	default:
+		return refuse()
+	}
+	return nil
+}

apps/backend/cmd/plugin-fixture/executor_provider_cleanup_checkpoint_test.go

@@ -0,0 +1,101 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/kdlbs/kandev/pluginsdk"
+	"github.com/stretchr/testify/require"
+)
+
+// checkpointingHost is a Host whose executor-provider callbacks record the
+// checkpoints a provider issues, refusing them when asked.
+type checkpointingHost struct {
+	pluginsdk.Host
+	checkpoints  []*pluginsdk.CheckpointExecutorResourceRequest
+	refuse       error
+	onCheckpoint func()
+}
+
+func (h *checkpointingHost) ExecutorProvider() pluginsdk.ExecutorProviderHostAPI { return h }
+
+func (h *checkpointingHost) CheckpointExecutorResource(_ context.Context, req *pluginsdk.CheckpointExecutorResourceRequest) (*pluginsdk.CheckpointExecutorResourceResponse, error) {
+	h.checkpoints = append(h.checkpoints, req)
+	if h.onCheckpoint != nil {
+		h.onCheckpoint()
+	}
+	if h.refuse != nil {
+		return nil, h.refuse
+	}
+	return &pluginsdk.CheckpointExecutorResourceResponse{Accepted: true}, nil
+}
+
+func (h *checkpointingHost) ReportExecutorProgress(context.Context, *pluginsdk.ReportExecutorProgressRequest) (*pluginsdk.ReportExecutorProgressResponse, error) {
+	return &pluginsdk.ReportExecutorProgressResponse{}, nil
+}
+
+func (h *checkpointingHost) ReadExecutorRuntimeArtifact(context.Context, *pluginsdk.ReadExecutorRuntimeArtifactRequest, func(*pluginsdk.ExecutorRuntimeArtifactChunk) error) error {
+	return nil
+}
+
+func provisionForCleanup(t *testing.T, plugin *fixturePlugin) (*pluginsdk.ExecutorProviderRequestContext, *pluginsdk.ExecutorResourceDescriptor) {
+	t.Helper()
+	profile := &pluginsdk.ExecutorProfileSnapshot{ProfileId: "profile-1", Config: map[string]string{"region": "eu-west-1"}}
+	ctx := &pluginsdk.ExecutorProviderRequestContext{
+		PluginId: "kandev-plugin-e2e", InstallationId: "installation-1", ProviderKey: executorProviderKey,
+		ContractVersion: 1, EnvironmentId: "environment-1", OperationId: "operation-1", InputDigest: "digest-1",
+	}
+	provisioned, err := plugin.ProvisionExecutorEnvironment(context.Background(), &pluginsdk.ProvisionExecutorEnvironmentRequest{Context: ctx, Profile: profile})
+	require.NoError(t, err)
+	return ctx, provisioned.GetResource()
+}
+
+// A destroy under a cleanup claim checkpoints cleanup_progress through the host
+// before it mutates anything, as a real provider must.
+func TestFixtureDestroyCheckpointsCleanupProgressBeforeMutating(t *testing.T) {
+	plugin := newFixturePluginAt(t.TempDir())
+	t.Cleanup(func() { require.NoError(t, plugin.Close()) })
+	ctx, resource := provisionForCleanup(t, plugin)
+	host := &checkpointingHost{}
+	plugin.SetHost(host)
+	host.onCheckpoint = func() {
+		_, found, err := plugin.fixtureResourceByHandle(resource.GetResourceHandle())
+		require.NoError(t, err)
+		require.True(t, found, "the resource was destroyed before the checkpoint")
+	}
+
+	destroyed, err := plugin.DestroyExecutorEnvironment(context.Background(), &pluginsdk.DestroyExecutorEnvironmentRequest{
+		Context: ctx, Resource: resource, CleanupClaim: "operation-1:cleanup",
+		CleanupReason: "launch_failed", DrainOutcome: "not_attempted",
+	})
+	require.NoError(t, err)
+	require.True(t, destroyed.GetConfirmedAbsent())
+	require.Len(t, host.checkpoints, 1)
+	require.Equal(t, "cleanup_progress", host.checkpoints[0].GetPhase())
+	require.EqualValues(t, 1, host.checkpoints[0].GetSequence())
+	require.Equal(t, resource.GetResourceHandle(), host.checkpoints[0].GetResource().GetResourceHandle())
+	_, found, err := plugin.fixtureResourceByHandle(resource.GetResourceHandle())
+	require.NoError(t, err)
+	require.False(t, found)
+}
+
+// A refused checkpoint fails the destroy closed: nothing is mutated and the
+// resource is not reported absent.
+func TestFixtureDestroyFailsClosedWhenTheHostRefusesTheCheckpoint(t *testing.T) {
+	plugin := newFixturePluginAt(t.TempDir())
+	t.Cleanup(func() { require.NoError(t, plugin.Close()) })
+	ctx, resource := provisionForCleanup(t, plugin)
+	plugin.SetHost(&checkpointingHost{refuse: errors.New("cleanup claim not found")})
+
+	destroyed, err := plugin.DestroyExecutorEnvironment(context.Background(), &pluginsdk.DestroyExecutorEnvironmentRequest{
+		Context: ctx, Resource: resource, CleanupClaim: "operation-1:cleanup",
+		CleanupReason: "launch_failed", DrainOutcome: "not_attempted",
+	})
+	require.NoError(t, err)
+	require.False(t, destroyed.GetConfirmedAbsent())
+	require.Equal(t, "cleanup_checkpoint_refused", destroyed.GetError().GetCode())
+	_, found, err := plugin.fixtureResourceByHandle(resource.GetResourceHandle())
+	require.NoError(t, err)
+	require.True(t, found, "the resource was mutated without an accepted checkpoint")
+}

apps/backend/cmd/plugin-fixture/executor_provider_drain_evidence_test.go

@@ -0,0 +1,74 @@
+package main
+
+import (
+	"context"
+	"testing"
+
+	"github.com/kdlbs/kandev/pluginsdk"
+	"github.com/stretchr/testify/require"
+)
+
+func destroyWithEvidence(t *testing.T, mutate func(*pluginsdk.DestroyExecutorEnvironmentRequest)) *pluginsdk.DestroyExecutorEnvironmentResponse {
+	t.Helper()
+	plugin := newFixturePluginAt(t.TempDir())
+	t.Cleanup(func() { require.NoError(t, plugin.Close()) })
+	ctx, resource := provisionForCleanup(t, plugin)
+	plugin.SetHost(&checkpointingHost{})
+	request := &pluginsdk.DestroyExecutorEnvironmentRequest{Context: ctx, Resource: resource, CleanupClaim: "claim-1", CleanupReason: "task_cleanup"}
+	mutate(request)
+	response, err := plugin.DestroyExecutorEnvironment(context.Background(), request)
+	require.NoError(t, err)
+	return response
+}
+
+// A provider that requires drain_receipt accepts not_attempted only as the
+// no-runtime proof of a failed launch, and a receipt only for its own claim.
+func TestFixtureDestroyAppliesTheDrainEvidenceRules(t *testing.T) {
+	receipt := func(claim string) *pluginsdk.ExecutorDrainReceipt {
+		return &pluginsdk.ExecutorDrainReceipt{ReceiptId: "receipt-1", CleanupClaim: claim, RuntimeIdentity: "runtime-1"}
+	}
+	cases := []struct {
+		name    string
+		mutate  func(*pluginsdk.DestroyExecutorEnvironmentRequest)
+		refused bool
+	}{
+		{"no outcome", func(*pluginsdk.DestroyExecutorEnvironmentRequest) {}, true},
+		{"not attempted under task cleanup", func(r *pluginsdk.DestroyExecutorEnvironmentRequest) { r.DrainOutcome = "not_attempted" }, true},
+		{"not attempted under launch failure", func(r *pluginsdk.DestroyExecutorEnvironmentRequest) {
+			r.DrainOutcome, r.CleanupReason = "not_attempted", "launch_failed"
+		}, false},
+		{"drained without a receipt", func(r *pluginsdk.DestroyExecutorEnvironmentRequest) { r.DrainOutcome = "drained" }, true},
+		{"drained with another claim's receipt", func(r *pluginsdk.DestroyExecutorEnvironmentRequest) {
+			r.DrainOutcome, r.DrainReceipt = "drained", receipt("claim-0")
+		}, true},
+		{"drained with this claim's receipt", func(r *pluginsdk.DestroyExecutorEnvironmentRequest) {
+			r.DrainOutcome, r.DrainReceipt = "drained", receipt("claim-1")
+		}, false},
+		{"uncertain", func(r *pluginsdk.DestroyExecutorEnvironmentRequest) { r.DrainOutcome = "uncertain" }, false},
+		{"unreachable", func(r *pluginsdk.DestroyExecutorEnvironmentRequest) { r.DrainOutcome = "unreachable" }, false},
+		{"uncertain with a receipt", func(r *pluginsdk.DestroyExecutorEnvironmentRequest) {
+			r.DrainOutcome, r.DrainReceipt = "uncertain", receipt("claim-1")
+		}, true},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			response := destroyWithEvidence(t, tc.mutate)
+			if tc.refused {
+				require.Equal(t, "drain_evidence_refused", response.GetError().GetCode())
+				require.False(t, response.GetConfirmedAbsent())
+				return
+			}
+			require.Nil(t, response.GetError())
+			require.True(t, response.GetConfirmedAbsent())
+		})
+	}
+}
+
+// A provider configured without the capabilities neither checkpoints nor
+// demands drain evidence.
+func TestFixtureWithoutDeclaredCapabilitiesAcceptsABareDestroy(t *testing.T) {
+	t.Setenv(fixtureCapabilitiesEnv, "")
+	response := destroyWithEvidence(t, func(*pluginsdk.DestroyExecutorEnvironmentRequest) {})
+	require.Nil(t, response.GetError())
+	require.True(t, response.GetConfirmedAbsent())
+}

apps/backend/cmd/plugin-fixture/executor_provider_test.go

@@ -99,6 +99,7 @@ func TestPluginExecutorFixtureContract(t *testing.T) {
 
 	destroyed, err := plugin.DestroyExecutorEnvironment(context.Background(), &pluginsdk.DestroyExecutorEnvironmentRequest{
 		Context: ctx, Resource: provisioned.GetResource(),
+		CleanupReason: "launch_failed", DrainOutcome: "not_attempted",
 	})
 	require.NoError(t, err)
 	require.True(t, destroyed.GetConfirmedAbsent())
@@ -179,13 +180,15 @@ func TestPluginExecutorFixtureFailureBarriersAreRetryable(t *testing.T) {
 	t.Setenv(fixtureExecutorBarrierEnv, fixtureBarrierUnconfirmedCleanupOnce)
 	delayed, err := plugin.DestroyExecutorEnvironment(context.Background(), &pluginsdk.DestroyExecutorEnvironmentRequest{
 		Context: ctx, Resource: provisioned.GetResource(),
+		CleanupReason: "launch_failed", DrainOutcome: "not_attempted",
 	})
 	require.NoError(t, err)
 	require.False(t, delayed.GetConfirmedAbsent())
 	require.Equal(t, "cleanup_unconfirmed", delayed.GetError().GetCode())
 
 	cleaned, err := plugin.DestroyExecutorEnvironment(context.Background(), &pluginsdk.DestroyExecutorEnvironmentRequest{
 		Context: ctx, Resource: provisioned.GetResource(),
+		CleanupReason: "launch_failed", DrainOutcome: "not_attempted",
 	})
 	require.NoError(t, err)
 	require.True(t, cleaned.GetConfirmedAbsent())
@@ -216,3 +219,26 @@ func fixtureHTTPSRequest(t *testing.T, plugin *fixturePlugin, lease *pluginsdk.E
 	require.NoError(t, err)
 	return string(body)
 }
+
+func TestPluginExecutorFixtureSameOperationSameInputReusesAndChangedInputConflicts(t *testing.T) {
+	plugin := newFixturePluginAt(t.TempDir())
+	t.Cleanup(func() { require.NoError(t, plugin.Close()) })
+	profile := &pluginsdk.ExecutorProfileSnapshot{ProfileId: "profile-1", Config: map[string]string{"region": "eu-west-1"}}
+	request := func(digest string) *pluginsdk.ProvisionExecutorEnvironmentRequest {
+		return &pluginsdk.ProvisionExecutorEnvironmentRequest{Profile: profile, Context: &pluginsdk.ExecutorProviderRequestContext{
+			PluginId: "kandev-plugin-e2e", InstallationId: "installation-1", ProviderKey: executorProviderKey,
+			ContractVersion: 1, EnvironmentId: "environment-1", OperationId: "operation-1", InputDigest: digest,
+		}}
+	}
+	_, err := plugin.ProvisionExecutorEnvironment(context.Background(), request("digest-intent-a"))
+	require.NoError(t, err)
+	_, err = plugin.ProvisionExecutorEnvironment(context.Background(), request("digest-intent-a"))
+	require.NoError(t, err)
+	require.Equal(t, 1, plugin.executorAllocationCount(), "a retry with the same intent must not allocate again")
+
+	_, err = plugin.ProvisionExecutorEnvironment(context.Background(), request("digest-intent-b"))
+	var actionErr *pluginsdk.ActionError
+	require.ErrorAs(t, err, &actionErr)
+	require.Equal(t, pluginsdk.ActionErrorConflict, actionErr.Code)
+	require.Equal(t, 1, plugin.executorAllocationCount())
+}

apps/backend/cmd/plugin-fixture/fixture-package/manifest.yaml

@@ -71,6 +71,9 @@ executor_providers:
         image:
           type: string
           title: Image
+          default: fixture-agent:latest
+          minLength: 3
+          maxLength: 32
         workspace_label:
           type: string
           title: Environment label with additional context for long names