[procmgr] Windows spawn profiles foundation

.github/CODEOWNERS

@@ -685,7 +685,7 @@
 /pkg/pidfile/                           @DataDog/agent-runtimes
 /pkg/persistentcache/                   @DataDog/agent-runtimes
 /pkg/procmgr/                           @DataDog/agent-runtimes
-/pkg/procmgr/rust/src/platform/windows.rs          @DataDog/agent-runtimes @DataDog/windows-products
+/pkg/procmgr/rust/src/platform/windows/             @DataDog/agent-runtimes @DataDog/windows-products
 /pkg/procmgr/rust/src/transport/named_pipe.rs       @DataDog/agent-runtimes @DataDog/windows-products
 /pkg/privateactionrunner/               @DataDog/action-platform
 /pkg/privateactionrunner/bundles/remoteaction/rshell  @DataDog/action-platform @DataDog/dd-agent-mcp @DataDog/fleet-remediation

pkg/procmgr/coat/BUILD.bazel

@@ -108,15 +108,30 @@ go_library(
             "@com_github_microsoft_go_winio//:go-winio",
             "@org_golang_google_grpc//:grpc",
             "@org_golang_google_grpc//credentials/insecure",
+            "@org_golang_x_sys//windows",
         ],
         "//conditions:default": [],
     }),
 )
 
 dd_agent_go_test(
     name = "coat_test",
+    # keep: split from coat_linux_test.
+    srcs = ["services_test.go"],
+    embed = [":coat"],
+    # keep: procmgr dep is in coat_linux_test only.
+    deps = [
+        "@com_github_stretchr_testify//assert",
+        "@com_github_stretchr_testify//require",
+    ],
+)
+
+dd_agent_go_test(
+    name = "coat_linux_test",
+    # keep: split from coat_test.
     srcs = ["collector_test.go"],
     embed = [":coat"],
+    target_compatible_with = ["@platforms//os:linux"],
     deps = [
         "//pkg/proto/pbgo/procmgr",
         "@com_github_stretchr_testify//assert",

pkg/procmgr/coat/client_grpc_windows.go

@@ -13,6 +13,7 @@ import (
 	"net"
 
 	"github.com/Microsoft/go-winio"
+	"golang.org/x/sys/windows"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 )
@@ -21,8 +22,13 @@ func dialProcmgrGRPC(socketPath string) (*grpc.ClientConn, error) {
 	conn, err := grpc.NewClient(
 		"passthrough:///procmgr",
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
-		grpc.WithContextDialer(func(_ context.Context, _ string) (net.Conn, error) {
-			return winio.DialPipe(socketPath, nil)
+		grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) {
+			return winio.DialPipeAccessImpLevel(
+				ctx,
+				socketPath,
+				uint32(windows.GENERIC_READ|windows.GENERIC_WRITE),
+				winio.PipeImpLevelIdentification,
+			)
 		}),
 	)
 	if err != nil {

pkg/procmgr/coat/collector_test.go

@@ -3,6 +3,8 @@
 // This product includes software developed at Datadog (https://www.datadoghq.com/).
 // Copyright 2026-present Datadog, Inc.
 
+//go:build linux
+
 package coat
 
 import (
@@ -59,9 +61,13 @@ func (s *mockSession) Disconnect() error {
 	return nil
 }
 
-func serviceSnapshotByID(t *testing.T, snapshot Snapshot, id string) ServiceSnapshot {
+func requireServiceCount(t *testing.T, snapshot Snapshot) {
 	t.Helper()
+	require.Len(t, snapshot.Services, len(migratableServices))
+}
 
+func serviceSnapshotByID(t *testing.T, snapshot Snapshot, id string) ServiceSnapshot {
+	t.Helper()
 	for _, service := range snapshot.Services {
 		if service.ID == id {
 			return service
@@ -79,43 +85,41 @@ func installMarkerForTest(t *testing.T, root string, service MigratableService,
 	return markers[index]
 }
 
-func setupDDOTInstallFixture(t *testing.T) string {
+func setupInstallFixture(t *testing.T, service MigratableService, markerRel string) string {
 	t.Helper()
 
-	ddot, ok := serviceByID("ddot")
-	require.True(t, ok)
-
 	root := t.TempDir()
-	marker := installMarkerForTest(t, root, ddot, 0)
+	marker := filepath.Join(root, markerRel)
 	require.NoError(t, os.MkdirAll(filepath.Dir(marker), 0o755))
 	require.NoError(t, os.WriteFile(marker, []byte("bin"), 0o644))
 	require.NoError(t, os.MkdirAll(filepath.Join(root, processesDirRel), 0o755))
 	require.NoError(t, os.WriteFile(
-		filepath.Join(root, processesDirRel, ddot.ProcmgrConfigFile),
+		filepath.Join(root, processesDirRel, service.ProcmgrConfigFile),
 		[]byte("cfg"),
 		0o644,
 	))
 	return root
 }
 
+func setupDDOTInstallFixture(t *testing.T) string {
+	t.Helper()
+
+	ddot, ok := serviceByID("ddot")
+	require.True(t, ok)
+	return setupInstallFixture(t, ddot, ddot.InstallMarkerRels[0])
+}
+
 func TestCollectInstalledViaStandaloneMarkerOnly(t *testing.T) {
 	ddot, ok := serviceByID("ddot")
 	require.True(t, ok)
 
-	root := t.TempDir()
-	standalone := installMarkerForTest(t, root, ddot, 1)
-	require.NoError(t, os.MkdirAll(filepath.Dir(standalone), 0o755))
-	require.NoError(t, os.WriteFile(standalone, []byte("bin"), 0o644))
-	require.NoError(t, os.MkdirAll(filepath.Join(root, processesDirRel), 0o755))
-	require.NoError(t, os.WriteFile(
-		filepath.Join(root, processesDirRel, ddot.ProcmgrConfigFile),
-		[]byte("cfg"),
-		0o644,
-	))
+	root := setupInstallFixture(t, ddot, ddot.InstallMarkerRels[1])
 
 	collector := NewCollectorWithClient(root, &mockClient{})
 
 	snapshot := collector.Collect(context.Background())
+	requireServiceCount(t, snapshot)
+
 	service := serviceSnapshotByID(t, snapshot, "ddot")
 	assert.True(t, service.Installed,
 		"standalone datadog-agent-ddot layout uses embedded/bin/otel-agent without ext/ddot")
@@ -132,6 +136,7 @@ func TestCollectServiceProcmgrRunning(t *testing.T) {
 	})
 
 	snapshot := collector.Collect(context.Background())
+	requireServiceCount(t, snapshot)
 
 	service := serviceSnapshotByID(t, snapshot, "ddot")
 	assert.Equal(t, "ddot", service.ID)
@@ -153,7 +158,7 @@ func TestCollectADPProcmgrRunning(t *testing.T) {
 	require.NoError(t, os.WriteFile(marker, []byte("bin"), 0o644))
 	require.NoError(t, os.MkdirAll(filepath.Join(root, processesDirRel), 0o755))
 	require.NoError(t, os.WriteFile(
-		filepath.Join(root, processesDirRel, "datadog-agent-data-plane.yaml"),
+		filepath.Join(root, processesDirRel, adp.ProcmgrConfigFile),
 		[]byte("cfg"),
 		0o644,
 	))
@@ -166,6 +171,7 @@ func TestCollectADPProcmgrRunning(t *testing.T) {
 	})
 
 	snapshot := collector.Collect(context.Background())
+	requireServiceCount(t, snapshot)
 
 	service := serviceSnapshotByID(t, snapshot, "agent-data-plane")
 	assert.Equal(t, "agent-data-plane", service.ID)
@@ -185,6 +191,7 @@ func TestCollectServiceProcmgrNotRunningStillManaged(t *testing.T) {
 	})
 
 	snapshot := collector.Collect(context.Background())
+	requireServiceCount(t, snapshot)
 
 	service := serviceSnapshotByID(t, snapshot, "ddot")
 	assert.Equal(t, ManagementModeProcmgr, service.ManagementMode)
@@ -197,12 +204,14 @@ func TestCollectNoProcmgrNoLegacy(t *testing.T) {
 	collector := NewCollectorWithClient(root, &mockClient{})
 
 	snapshot := collector.Collect(context.Background())
+	requireServiceCount(t, snapshot)
 
-	service := serviceSnapshotByID(t, snapshot, "ddot")
-	assert.False(t, service.Installed)
-	assert.False(t, service.ProcmgrConfigured)
-	assert.Equal(t, ManagementModeNone, service.ManagementMode)
-	assert.Equal(t, pb.ProcessState_UNKNOWN, service.ProcmgrState)
+	for _, service := range snapshot.Services {
+		assert.False(t, service.Installed)
+		assert.False(t, service.ProcmgrConfigured)
+		assert.Equal(t, ManagementModeNone, service.ManagementMode)
+		assert.Equal(t, pb.ProcessState_UNKNOWN, service.ProcmgrState)
+	}
 }
 
 func TestCollectInstallMarkerAbsent(t *testing.T) {
@@ -220,6 +229,7 @@ func TestCollectInstallMarkerAbsent(t *testing.T) {
 	collector := NewCollectorWithClient(root, &mockClient{})
 
 	snapshot := collector.Collect(context.Background())
+	requireServiceCount(t, snapshot)
 
 	service := serviceSnapshotByID(t, snapshot, "ddot")
 	assert.False(t, service.Installed, "without install marker, Installed must stay false")
@@ -239,6 +249,7 @@ func TestCollectProcmgrConfigAbsent(t *testing.T) {
 	collector := NewCollectorWithClient(root, &mockClient{})
 
 	snapshot := collector.Collect(context.Background())
+	requireServiceCount(t, snapshot)
 
 	service := serviceSnapshotByID(t, snapshot, "ddot")
 	assert.True(t, service.Installed)
@@ -260,6 +271,8 @@ func TestCollectDaemonUnreachable(t *testing.T) {
 
 	assert.False(t, snapshot.Daemon.Reachable, "daemon status error should yield empty snapshot")
 	assert.False(t, snapshot.Daemon.Ready)
+	requireServiceCount(t, snapshot)
+
 	service := serviceSnapshotByID(t, snapshot, "ddot")
 	assert.Equal(t, ManagementModeNone, service.ManagementMode,
 		"daemon failure prevents listing processes")
@@ -279,6 +292,8 @@ func TestCollectDaemonReachableListFails(t *testing.T) {
 
 	assert.True(t, snapshot.Daemon.Reachable)
 	assert.True(t, snapshot.Daemon.Ready)
+	requireServiceCount(t, snapshot)
+
 	service := serviceSnapshotByID(t, snapshot, "ddot")
 	assert.Equal(t, ManagementModeNone, service.ManagementMode)
 	assert.Equal(t, pb.ProcessState_UNKNOWN, service.ProcmgrState)

pkg/procmgr/coat/services_test.go

@@ -0,0 +1,35 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2026-present Datadog, Inc.
+
+package coat
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMigratableServicesCatalog(t *testing.T) {
+	require.NotEmpty(t, migratableServices)
+
+	seen := make(map[string]struct{}, len(migratableServices))
+	for _, service := range migratableServices {
+		_, dup := seen[service.ID]
+		assert.False(t, dup, "duplicate service id %q", service.ID)
+		seen[service.ID] = struct{}{}
+
+		assert.NotEmpty(t, service.ProcmgrProcessName)
+		assert.NotEmpty(t, service.ProcmgrConfigFile)
+		assert.NotEmpty(t, service.InstallMarkerRels)
+		assert.NotEmpty(t, service.LegacySystemdUnits)
+	}
+}
+
+func TestServiceByID(t *testing.T) {
+	service, ok := serviceByID("ddot")
+	require.True(t, ok)
+	assert.Equal(t, "datadog-agent-ddot", service.ProcmgrProcessName)
+}

pkg/procmgr/rust/Cargo.toml

@@ -37,7 +37,7 @@ tonic-reflection.workspace = true
 uuid.workspace = true
 
 [target.'cfg(unix)'.dependencies]
-nix = { workspace = true, features = ["signal", "process"] }
+nix = { workspace = true, features = ["signal", "process", "user"] }
 
 [target.'cfg(windows)'.dependencies]
 windows-registry.workspace = true
@@ -48,9 +48,16 @@ windows-sys = { workspace = true, features = [
     "Win32_System_Threading",
     "Win32_System_Console",
     "Win32_System_Environment",
+    "Win32_UI_Shell",
     "Win32_Security",
+    "Win32_Security_Authentication_Identity",
+    "Win32_Security_Authorization",
+    "Win32_System_Pipes",
     "Win32_System_Registry",
+    "Win32_System_SystemInformation",
+    "Win32_System_WindowsProgramming",
     "Win32_Storage_FileSystem",
+    "Win32_NetworkManagement_NetManagement",
 ] }
 
 [target.'cfg(unix)'.dev-dependencies]

pkg/procmgr/rust/src/bins/dd-procmgrd.rs

@@ -19,7 +19,7 @@ fn main() {
 
 #[cfg(windows)]
 fn run() -> anyhow::Result<()> {
-    dd_procmgrd::service::run_as_service()
+    dd_procmgrd::platform::run_as_service()
 }
 
 #[cfg(not(windows))]
@@ -41,6 +41,6 @@ fn run() -> anyhow::Result<()> {
     runtime.block_on(async {
         let loader = Arc::new(YamlConfigLoader::from_env());
         let mgr = ProcessManager::new(loader, Arc::new(V4UuidGenerator));
-        mgr.run().await
+        mgr.supervisor().run().await
     })
 }

pkg/procmgr/rust/src/env.rs

@@ -4,6 +4,42 @@
 // Copyright 2026-present Datadog, Inc.
 
 use anyhow::{Context, Result};
+use log::warn;
+
+pub(crate) fn expand_env_vars(input: &str) -> String {
+    expand_vars_with(input, |name| std::env::var(name).ok())
+}
+
+pub(crate) fn expand_vars_with(input: &str, lookup: impl Fn(&str) -> Option<String>) -> String {
+    let mut out = String::with_capacity(input.len());
+    let mut rest = input;
+    while let Some(start) = rest.find("${") {
+        out.push_str(&rest[..start]);
+        let after = &rest[start + 2..];
+        match after.find('}') {
+            Some(end) => {
+                let name = &after[..end];
+                match lookup(name) {
+                    Some(val) => out.push_str(&val),
+                    None => {
+                        warn!(
+                            "process config references unset variable ${{{name}}}, leaving it literal"
+                        );
+                        out.push_str(&rest[start..start + 2 + end + 1]);
+                    }
+                }
+                rest = &after[end + 1..];
+            }
+            None => {
+                // No closing brace: emit verbatim.
+                out.push_str(&rest[start..]);
+                return out;
+            }
+        }
+    }
+    out.push_str(rest);
+    out
+}
 
 /// Parse a systemd-style environment file into key-value pairs.
 /// Supports `KEY=VALUE`, `KEY="VALUE"`, `KEY='VALUE'`, comments (#), and blank lines.

pkg/procmgr/rust/src/grpc/caller_auth.rs

@@ -0,0 +1,30 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2026-present Datadog, Inc.
+
+use tonic::{Request, Status};
+
+#[cfg(windows)]
+use crate::transport::PipeCallerAuth;
+
+/// Windows: `Create` requires Administrator or LocalSystem pipe client.
+#[cfg(windows)]
+pub(crate) fn require_privileged_pipe_client<T>(request: &Request<T>) -> Result<(), Status> {
+    let may_mutate = request
+        .extensions()
+        .get::<PipeCallerAuth>()
+        .map(PipeCallerAuth::may_mutate)
+        .unwrap_or(false);
+    if !may_mutate {
+        return Err(Status::permission_denied(
+            "operation requires an Administrator or LocalSystem pipe client",
+        ));
+    }
+    Ok(())
+}
+
+#[cfg(not(windows))]
+pub(crate) fn require_privileged_pipe_client<T>(_request: &Request<T>) -> Result<(), Status> {
+    Ok(())
+}