feat: target-state worker secrets via TaskDef.runtimeMetadata

sdk/csharp/src/Conductor.AI/Conductor.AI.csproj

@@ -38,6 +38,9 @@
   <ItemGroup>
     <!-- System.Text.Json and System.Threading.Channels are in-box on .NET 10 -->
     <!-- conductor-csharp brings Newtonsoft.Json + Microsoft.Extensions.Logging.Abstractions transitively -->
+    <!-- TARGET: requires a conductor-csharp release that carries Task.RuntimeMetadata (host-resolved
+         worker secrets, wire-only; conductor-oss/csharp-sdk feat/task-runtime-metadata). The field is
+         additive; repin to that release once it lands. Not built here (no dotnet in the authoring env). -->
     <PackageReference Include="conductor-csharp" Version="1.1.4" />
     <PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
   </ItemGroup>

sdk/csharp/src/Conductor.AI/WorkerManager.cs

@@ -95,20 +95,20 @@ private async System.Threading.Tasks.Task ExecuteAsync(Task task, CancellationTo
 
             // Strip internal keys from the handler-visible input
             var handlerInput = inputData
-                .Where(kv => !string.Equals(kv.Key, "__agentspan_ctx__",       StringComparison.OrdinalIgnoreCase)
-                          && !string.Equals(kv.Key, "_agent_state",            StringComparison.OrdinalIgnoreCase)
-                          && !string.Equals(kv.Key, "__resolved_credentials__", StringComparison.OrdinalIgnoreCase)
-                          && !string.Equals(kv.Key, "method",                  StringComparison.OrdinalIgnoreCase))
+                .Where(kv => !string.Equals(kv.Key, "__agentspan_ctx__", StringComparison.OrdinalIgnoreCase)
+                          && !string.Equals(kv.Key, "_agent_state",      StringComparison.OrdinalIgnoreCase)
+                          && !string.Equals(kv.Key, "method",            StringComparison.OrdinalIgnoreCase))
                 .ToDictionary(kv => kv.Key, kv => kv.Value, StringComparer.OrdinalIgnoreCase);
 
             // Resolve and inject credentials via the centralized helper so the
             // mutation + invocation + restoration is atomic under a single
             // process-wide lock. See docs/design/secret-injection-contract.md.
             // Tier-2 (env-injection) path; tier-1 (explicit-key) lands when the
             // user-facing API exposes a `credentials` parameter to agent factories.
-            // Embedded: the host resolves ${workflow.secrets.NAME} into __resolved_credentials__
-            // at poll time. Prefer that map; otherwise fall back to the native token-pull.
-            var resolvedCredentials = ReadResolvedCredentials(inputData);
+            // Embedded: the host resolves the worker's declared TaskDef.runtimeMetadata secret names
+            // at poll time and delivers the values on the wire-only Task.RuntimeMetadata (never
+            // persisted). Prefer that map; otherwise fall back to the native token-pull.
+            var resolvedCredentials = ReadRuntimeMetadata(task);
             if (resolvedCredentials.Count == 0 && _credentialNames.Length > 0)
             {
                 var creds = await _http.ResolveCredentialsAsync(
@@ -221,18 +221,19 @@ or CredentialRateLimitException
     }
 
     /// <summary>
-    /// Read the host-delivered <c>__resolved_credentials__</c> name→value map from task input
-    /// (embedded mode). The host resolves the stamped <c>${workflow.secrets.NAME}</c> references at
-    /// poll time. Empty when absent (standalone → the native token-pull is used instead).
+    /// Read the host-delivered secret name→value map from <c>Task.RuntimeMetadata</c> (embedded
+    /// mode). The host resolves the worker's declared <c>TaskDef.runtimeMetadata</c> names from its
+    /// secret store at poll time and injects the values on the wire only — never persisted to task
+    /// input (conductor-oss PR #1255). Empty when absent (standalone → the native token-pull).
     /// </summary>
-    private static Dictionary<string, string> ReadResolvedCredentials(Dictionary<string, JsonElement> inputData)
+    private static Dictionary<string, string> ReadRuntimeMetadata(Task task)
     {
         var result = new Dictionary<string, string>();
-        if (inputData.TryGetValue("__resolved_credentials__", out var rc) && rc.ValueKind == JsonValueKind.Object)
+        if (task?.RuntimeMetadata is { Count: > 0 } rm)
         {
-            foreach (var prop in rc.EnumerateObject())
-                if (prop.Value.ValueKind == JsonValueKind.String)
-                    result[prop.Name] = prop.Value.GetString()!;
+            foreach (var (k, v) in rm)
+                if (k is not null && v is not null)
+                    result[k] = v;
         }
         return result;
     }

sdk/csharp/tests/Conductor.AI.Tests/RuntimeMetadataReadTests.cs

@@ -0,0 +1,55 @@
+// Copyright (c) 2025 Agentspan
+// Licensed under the MIT License.
+
+using System.Collections.Generic;
+using System.Reflection;
+using Xunit;
+using ModelTask = Conductor.Client.Models.Task;
+
+namespace Conductor.AI.Tests;
+
+/// <summary>
+/// Embedded host-delivery read-path: the worker reads host-resolved secret values from
+/// <c>Task.RuntimeMetadata</c> (wire-only, resolved by the host from the worker's declared
+/// <c>TaskDef.runtimeMetadata</c>; conductor-oss PR #1255). Absent/empty yields an empty map
+/// (standalone falls back to the native token-pull).
+/// </summary>
+public class RuntimeMetadataReadTests
+{
+    private static Dictionary<string, string> Invoke(ModelTask task)
+    {
+        // WorkerPollLoop is internal; reach ReadRuntimeMetadata (private static) via reflection.
+        var type = typeof(CredentialScope).Assembly.GetType("Conductor.AI.WorkerPollLoop")!;
+        var method = type.GetMethod(
+            "ReadRuntimeMetadata",
+            BindingFlags.NonPublic | BindingFlags.Static)!;
+        return (Dictionary<string, string>)method.Invoke(null, new object?[] { task })!;
+    }
+
+    [Fact]
+    public void Extracts_host_delivered_values()
+    {
+        var task = new ModelTask(
+            taskId: "t1",
+            runtimeMetadata: new Dictionary<string, string>
+            {
+                ["GITHUB_TOKEN"] = "ghp_host",
+                ["GH_APP_ID"] = "42",
+            });
+
+        var result = Invoke(task);
+
+        Assert.Equal(2, result.Count);
+        Assert.Equal("ghp_host", result["GITHUB_TOKEN"]);
+        Assert.Equal("42", result["GH_APP_ID"]);
+    }
+
+    [Fact]
+    public void Empty_when_absent_or_empty()
+    {
+        Assert.Empty(Invoke(new ModelTask(taskId: "t1")));
+        Assert.Empty(Invoke(new ModelTask(
+            taskId: "t1",
+            runtimeMetadata: new Dictionary<string, string>())));
+    }
+}

sdk/java/build.gradle

@@ -14,6 +14,9 @@ java {
 }
 
 repositories {
+    // TARGET: consumes the local conductor-client build that carries Task.runtimeMetadata
+    // (conductor-oss/java-sdk feat/task-runtime-metadata). Remove once that release lands.
+    mavenLocal()
     mavenCentral()
 }
 
@@ -28,7 +31,9 @@ ext {
     // separately from the server engine (engine = 3.30.2); wire-compatible with
     // the 3.x task REST API, bundles the common DTOs, and provides native auth
     // via io.orkes.conductor.client.ApiClient (key/secret → token).
-    conductorClientVersion = '5.0.1'
+    // TARGET: 5.1.0 adds Task.runtimeMetadata (host-resolved worker secrets, wire-only).
+    // Currently a local mavenLocal build; repin to the published 5.1.0 once it releases.
+    conductorClientVersion = '5.1.0'
 }
 
 dependencies {

sdk/java/src/main/java/org/conductoross/conductor/ai/internal/WorkerManager.java

@@ -250,7 +250,23 @@ public void register(
         logger.info("Registered worker for task: {} (domain={})", taskName, domain);
     }
 
-    private void registerTaskDef(String taskName, int configuredTimeoutSeconds) {
+    /**
+     * Register the worker TaskDef create-only: create it when absent, but never overwrite one that
+     * already exists. When embedded, the host server pre-registers the worker TaskDef and declares
+     * its secret names on {@code TaskDef.runtimeMetadata} (conductor-oss PR #1255); overwriting here
+     * with a bare def (the client TaskDef model carries no runtimeMetadata) would clobber that and
+     * starve the host resolver. Standalone still gets the def created when absent. The existence
+     * check chooses correctly with no embedded flag.
+     */
+    void registerTaskDef(String taskName, int configuredTimeoutSeconds) {
+        try {
+            if (metadataClient.getTaskDef(taskName) != null) {
+                logger.debug("Task def {} already exists — leaving it untouched (create-only)", taskName);
+                return;
+            }
+        } catch (Exception lookupFailed) {
+            // Not found (or lookup errored) — fall through and create it.
+        }
         try {
             long timeout = effectiveTaskTimeout(configuredTimeoutSeconds);
             TaskDef taskDef = new TaskDef(taskName);
@@ -373,9 +389,10 @@ private TaskResult executeHandler(String taskName, Task task) {
         // problem. See docs/design/secret-injection-contract.md.
         Map<String, String> resolvedSecrets = Collections.emptyMap();
         List<String> declared = taskCredentials.getOrDefault(taskName, Collections.emptyList());
-        // Embedded: the host resolves ${workflow.secrets.NAME} into __resolved_credentials__ at
-        // poll time. Prefer that map; otherwise fall back to the native token-pull (standalone).
-        Map<String, String> hostDelivered = readResolvedCredentials(inputData);
+        // Embedded: the host resolves the worker's declared TaskDef.runtimeMetadata secret names at
+        // poll time and delivers the values on the wire-only Task.runtimeMetadata (never persisted).
+        // Prefer that map; otherwise fall back to the native token-pull (standalone).
+        Map<String, String> hostDelivered = readRuntimeMetadata(task);
         if (!hostDelivered.isEmpty()) {
             resolvedSecrets = hostDelivered;
         } else if (!declared.isEmpty()) {
@@ -423,18 +440,19 @@ private TaskResult executeHandler(String taskName, Task task) {
     }
 
     /**
-     * Read the host-delivered {@code __resolved_credentials__} name→value map from task input
-     * (embedded mode). The host resolves the stamped {@code ${workflow.secrets.NAME}} references at
-     * poll time. Returns an empty map when absent (standalone → native token-pull is used instead).
+     * Read the host-delivered secret name→value map from {@code Task.runtimeMetadata} (embedded mode).
+     * The host resolves the worker's declared {@code TaskDef.runtimeMetadata} names from its secret
+     * store at poll time and injects the values on the wire only — never persisted to task input
+     * (conductor-oss PR #1255). Returns an empty map when absent (standalone → native token-pull).
      */
-    private static Map<String, String> readResolvedCredentials(Map<String, Object> inputData) {
-        if (inputData == null) return Collections.emptyMap();
-        Object rc = inputData.get("__resolved_credentials__");
-        if (!(rc instanceof Map<?, ?> m) || m.isEmpty()) return Collections.emptyMap();
+    private static Map<String, String> readRuntimeMetadata(Task task) {
+        if (task == null) return Collections.emptyMap();
+        Map<String, String> rm = task.getRuntimeMetadata();
+        if (rm == null || rm.isEmpty()) return Collections.emptyMap();
         Map<String, String> out = new HashMap<>();
-        for (Map.Entry<?, ?> e : m.entrySet()) {
-            if (e.getKey() != null && e.getValue() instanceof String s) {
-                out.put(e.getKey().toString(), s);
+        for (Map.Entry<String, String> e : rm.entrySet()) {
+            if (e.getKey() != null && e.getValue() != null) {
+                out.put(e.getKey(), e.getValue());
             }
         }
         return out;

sdk/java/src/test/java/org/conductoross/conductor/ai/SerializerTest.java

@@ -483,8 +483,10 @@ void llm_guardrail_requires_model_and_policy() {
     @Test
     @SuppressWarnings("unchecked")
     void on_condition_handoff_serialized_with_target() {
-        Agent supervisor =
-                Agent.builder().name("supervisor").model("anthropic/claude-sonnet-4-6").build();
+        Agent supervisor = Agent.builder()
+                .name("supervisor")
+                .model("anthropic/claude-sonnet-4-6")
+                .build();
         Agent worker = Agent.builder()
                 .name("worker")
                 .model("anthropic/claude-sonnet-4-6")
@@ -847,8 +849,10 @@ void planner_context_emitted_with_text_and_url_entries() {
         // Mirrors the Python + TS serializer tests. The wire shape MUST be
         // byte-equal across SDKs so the server compiler sees the same
         // payload regardless of language.
-        Agent planner =
-                Agent.builder().name("planner_sub").model("anthropic/claude-sonnet-4-6").build();
+        Agent planner = Agent.builder()
+                .name("planner_sub")
+                .model("anthropic/claude-sonnet-4-6")
+                .build();
         ToolDef stub = ToolDef.builder()
                 .name("stub")
                 .description("stub")
@@ -885,8 +889,10 @@ void planner_context_emitted_with_text_and_url_entries() {
     void planner_context_omitted_when_unset() {
         // Counterfactual: without plannerContext the field MUST NOT appear
         // on the wire. Pairs with the positive test — pins the gating.
-        Agent planner =
-                Agent.builder().name("planner_sub").model("anthropic/claude-sonnet-4-6").build();
+        Agent planner = Agent.builder()
+                .name("planner_sub")
+                .model("anthropic/claude-sonnet-4-6")
+                .build();
         ToolDef stub = ToolDef.builder()
                 .name("stub")
                 .description("stub")
@@ -907,7 +913,8 @@ void planner_context_omitted_when_unset() {
     void planner_context_rejected_on_non_plan_execute_strategy() {
         // Same guard shape as planner=/fallback= — setting plannerContext
         // on anything other than PLAN_EXECUTE is a silent bug.
-        Agent sub = Agent.builder().name("sub").model("anthropic/claude-sonnet-4-6").build();
+        Agent sub =
+                Agent.builder().name("sub").model("anthropic/claude-sonnet-4-6").build();
         IllegalArgumentException e = assertThrows(IllegalArgumentException.class, () -> Agent.builder()
                 .name("h")
                 .model("anthropic/claude-sonnet-4-6")
@@ -956,8 +963,10 @@ void parity_fields_serialized() {
 
     @Test
     void parity_fields_absent_when_unset() {
-        Agent agent =
-                Agent.builder().name("plain_agent").model("anthropic/claude-sonnet-4-6").build();
+        Agent agent = Agent.builder()
+                .name("plain_agent")
+                .model("anthropic/claude-sonnet-4-6")
+                .build();
         Map<String, Object> out = ser.serialize(agent);
         assertFalse(out.containsKey("reasoningEffort"), "reasoningEffort omitted when unset");
         assertFalse(out.containsKey("maskedFields"), "maskedFields omitted when unset");

sdk/java/src/test/java/org/conductoross/conductor/ai/internal/EmbeddedTaskDefRegistrationTest.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright (c) 2025 AgentSpan
+ * Licensed under the MIT License.
+ */
+package org.conductoross.conductor.ai.internal;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.lang.reflect.Field;
+import java.util.List;
+
+import org.conductoross.conductor.ai.AgentConfig;
+import org.junit.jupiter.api.Test;
+
+import com.netflix.conductor.client.http.ConductorClient;
+import com.netflix.conductor.client.http.MetadataClient;
+import com.netflix.conductor.common.metadata.tasks.TaskDef;
+
+/**
+ * Worker TaskDefs are registered create-only: the SDK creates the def when absent but never
+ * overwrites one that already exists. When embedded, the host server pre-registers the worker
+ * TaskDef and declares its secret names on TaskDef.runtimeMetadata (conductor-oss PR #1255);
+ * overwriting here with a bare def (the client TaskDef model has no runtimeMetadata field) would
+ * clobber that and starve the host resolver. No embedded flag — the existence check decides.
+ */
+class EmbeddedTaskDefRegistrationTest {
+
+    /** Fake client: reports whether a def "exists" and records any registration, without network. */
+    private static final class RecordingMetadataClient extends MetadataClient {
+        private final boolean exists;
+        boolean registered = false;
+
+        RecordingMetadataClient(boolean exists) {
+            this.exists = exists;
+        }
+
+        @Override
+        public TaskDef getTaskDef(String taskType) {
+            return exists ? new TaskDef(taskType) : null;
+        }
+
+        @Override
+        public void registerTaskDefs(List<TaskDef> taskDefs) {
+            this.registered = true;
+        }
+    }
+
+    private static boolean didRegister(boolean alreadyExists) throws Exception {
+        WorkerManager wm = new WorkerManager(new AgentConfig(), new ConductorClient());
+        RecordingMetadataClient client = new RecordingMetadataClient(alreadyExists);
+        Field f = WorkerManager.class.getDeclaredField("metadataClient");
+        f.setAccessible(true);
+        f.set(wm, client);
+        wm.registerTaskDef("check_secret", 300);
+        return client.registered;
+    }
+
+    @Test
+    void doesNotOverwriteExistingTaskDef() throws Exception {
+        // Existing def (e.g. server-registered with runtimeMetadata) must be left untouched.
+        assertFalse(didRegister(true), "must not overwrite an existing TaskDef");
+    }
+
+    @Test
+    void createsTaskDefWhenAbsent() throws Exception {
+        assertTrue(didRegister(false), "must create the TaskDef when none exists");
+    }
+}