Per-invocation properties: one resident instance, many callers

SW.Serverless.Contract/Protos/adapter.proto

@@ -47,6 +47,15 @@ message Invoke {
     // request. Traxis calls a command and then GetLogs; both must see the SAME session or the
     // audit trail comes back empty. Absent means "this call alone".
     string session_id = 4;
+
+    // Configuration for THIS call, on top of the startup values the process was given.
+    //
+    // An exclusive resident instance is shared: in Bitween one database connection serves every
+    // subscription bound to it, and each of those has its own settings — which statement to run,
+    // which operation it is. Startup values cannot carry that, because they belong to the process
+    // and the process belongs to all of them. This is the same reason session_id exists: a shared
+    // instance has to be told whose call this is.
+    map<string, string> properties = 5;
 }
 
 message Ping {}

SW.Serverless.Samples.Ticker/Handler.cs

@@ -156,6 +156,30 @@ public async Task<object> ClearCursor()
             return new { cleared = true };
         }
 
+        /// <summary>
+        /// Per-call configuration, which is what a SHARED instance needs: one process serving many
+        /// callers cannot take per-caller settings from its startup values, because those belong to
+        /// the process.
+        /// </summary>
+        public Task<object> ReadValue(string name) => Task.FromResult<object>(new
+        {
+            invocation = context.InvocationValues.TryGetValue(name, out var v) ? v : null,
+            startup = context.StartupValueOf(name),
+            resolved = context.ValueOf(name)
+        });
+
+        /// <summary>
+        /// Reads the same value either side of an await, so a test can prove two concurrent callers
+        /// do not see each other's — the failure an ordinary field would have.
+        /// </summary>
+        public async Task<object> ReadValueSlowly(string name)
+        {
+            var before = context.ValueOf(name);
+            await Task.Delay(300);
+            var after = context.ValueOf(name);
+            return new { before, after };
+        }
+
         /// <summary>Demonstrates that a command failure comes back as a typed error, not a hang.</summary>
         public Task Explode() => throw new InvalidOperationException("Deliberate failure from the ticker sample.");
 

SW.Serverless.Sdk/Resident/IAdapterContext.cs

@@ -28,6 +28,24 @@ public interface IAdapterContext
         IReadOnlyDictionary<string, string> AdapterValues { get; }
         string StartupValueOf(string name);
 
+        /// <summary>
+        /// Configuration for the command CURRENTLY running on this async flow, sent by the host
+        /// with the invocation. Empty outside a command, and empty when the caller sent none.
+        ///
+        /// It exists because an exclusive instance is shared. One database connection serves every
+        /// subscription pointed at it, and each of those has its own settings — which statement to
+        /// run, which operation it is. Startup values cannot carry that: they belong to the process,
+        /// and the process belongs to all of them.
+        /// </summary>
+        IReadOnlyDictionary<string, string> InvocationValues { get; }
+
+        /// <summary>
+        /// <see cref="InvocationValues"/> first, then <see cref="StartupValues"/>. This is the one
+        /// to reach for: a per-call setting overrides the process default, and an adapter serving a
+        /// single caller behaves exactly as it did before any of this existed.
+        /// </summary>
+        string ValueOf(string name);
+
         /// <summary>Cancelled when the host asks the adapter to shut down.</summary>
         CancellationToken Stopping { get; }
 

SW.Serverless.Sdk/Resident/ResidentRunner.cs

@@ -62,6 +62,11 @@ public sealed class ResidentRunner : IAdapterContext
         IReadOnlyDictionary<string, string> startupValues = new Dictionary<string, string>();
         IReadOnlyDictionary<string, string> adapterValues = new Dictionary<string, string>();
 
+        static readonly IReadOnlyDictionary<string, string> Empty = new Dictionary<string, string>();
+
+        // Per async flow, so concurrent commands on one shared instance do not read each other's.
+        static readonly AsyncLocal<IReadOnlyDictionary<string, string>> invocationValues = new();
+
         ResidentRunner(Type handlerType, Func<IAdapterContext, object> handlerFactory)
         {
             this.handlerType = handlerType;
@@ -304,6 +309,14 @@ async Task OnInvokeAsync(long id, Invoke invoke)
                 if (!commands.TryGetValue(invoke.Command, out var method))
                     throw new MissingMethodException(handlerType.FullName, invoke.Command);
 
+                // Set on THIS async flow, before the handler runs, so concurrent invocations on a
+                // shared instance each see their own caller's configuration. AsyncLocal rather than
+                // a field for exactly that reason: several commands are in flight at once by
+                // design, and a field would have the last one in overwrite the rest.
+                invocationValues.Value = invoke.Properties.Count == 0
+                    ? Empty
+                    : new Dictionary<string, string>(invoke.Properties);
+
                 // The host's session id when it grouped this call with others, otherwise the
                 // call stands alone.
                 using var session = AdapterSession.Begin(
@@ -488,6 +501,16 @@ async Task PumpOutboundAsync(IClientStreamWriter<AdapterFrame> stream)
         public string StartupValueOf(string name) =>
             startupValues.TryGetValue(name, out var v) ? v : null;
 
+        public IReadOnlyDictionary<string, string> InvocationValues => invocationValues.Value ?? Empty;
+
+        public string ValueOf(string name)
+        {
+            var perCall = invocationValues.Value;
+            if (perCall != null && perCall.TryGetValue(name, out var value)) return value;
+
+            return StartupValueOf(name);
+        }
+
         public async Task<PublishResult> PublishAsync(
             ReadOnlyMemory<byte> payload, string dedupeKey, string endpoint = null,
             IDictionary<string, string> headers = null, string contentType = null,

SW.Serverless.UnitTests/ResidentAdapterTests.cs

@@ -295,6 +295,84 @@ await WaitFor(() => adapters.Describe()
 
         // ------------------------------------------------------------------ helpers
 
+        // ------------------------------------------------------------------ per-invocation properties
+
+        /// <summary>
+        /// Configuration that travels with the CALL rather than with the process. An exclusive
+        /// instance is shared by every caller pointed at it, so anything that varies per caller —
+        /// which statement to run, which tenant this is — cannot live in the startup values.
+        /// </summary>
+        [TestMethod]
+        public async Task Per_invocation_properties_reach_the_command()
+        {
+            var instance = await StartTicker("percall");
+
+            var answer = await instance.InvokeAsync<Dictionary<string, string>>(
+                "ReadValue", "Tag",
+                properties: new Dictionary<string, string> { ["Tag"] = "invoice-run" });
+
+            Assert.AreEqual("invoice-run", answer["invocation"]);
+            Assert.IsNull(answer["startup"], "the process was never told about Tag");
+            Assert.AreEqual("invoice-run", answer["resolved"]);
+
+            await adapters.StopAsync(TickerId, "percall", drain: false);
+        }
+
+        /// <summary>
+        /// A per-call value wins over the process default, and an adapter called WITHOUT properties
+        /// still sees the startup value — which is what keeps every existing adapter working.
+        /// </summary>
+        [TestMethod]
+        public async Task A_per_invocation_value_overrides_the_startup_value()
+        {
+            var instance = await StartTicker("override", intervalSeconds: 30);
+
+            var overridden = await instance.InvokeAsync<Dictionary<string, string>>(
+                "ReadValue", "IntervalSeconds",
+                properties: new Dictionary<string, string> { ["IntervalSeconds"] = "5" });
+
+            Assert.AreEqual("5", overridden["resolved"]);
+            Assert.AreEqual("30", overridden["startup"], "the startup value is still there underneath");
+
+            var plain = await instance.InvokeAsync<Dictionary<string, string>>(
+                "ReadValue", "IntervalSeconds");
+
+            Assert.AreEqual("30", plain["resolved"], "no properties means the process default, as before");
+            Assert.IsNull(plain["invocation"]);
+
+            await adapters.StopAsync(TickerId, "override", drain: false);
+        }
+
+        /// <summary>
+        /// The reason this is an AsyncLocal and not a field. Several commands run at once on one
+        /// instance by design — that is what multiplexing is for — and a field would have the last
+        /// caller in overwrite everyone else's configuration mid-flight.
+        /// </summary>
+        [TestMethod]
+        public async Task Concurrent_invocations_do_not_see_each_others_properties()
+        {
+            var instance = await StartTicker("concurrent");
+
+            var first = instance.InvokeAsync<Dictionary<string, string>>(
+                "ReadValueSlowly", "Tag",
+                properties: new Dictionary<string, string> { ["Tag"] = "alpha" });
+
+            var second = instance.InvokeAsync<Dictionary<string, string>>(
+                "ReadValueSlowly", "Tag",
+                properties: new Dictionary<string, string> { ["Tag"] = "beta" });
+
+            await Task.WhenAll(first, second);
+
+            // Both sides of the await, because the value has to survive the continuation as well as
+            // arrive on the way in.
+            Assert.AreEqual("alpha", first.Result["before"]);
+            Assert.AreEqual("alpha", first.Result["after"]);
+            Assert.AreEqual("beta", second.Result["before"]);
+            Assert.AreEqual("beta", second.Result["after"]);
+
+            await adapters.StopAsync(TickerId, "concurrent", drain: false);
+        }
+
         // ------------------------------------------------------------------ host-held state
 
         /// <summary>

SW.Serverless/Resident/AdapterPool.cs

@@ -195,8 +195,10 @@ public Lease(AdapterPool pool, ResidentAdapterInstance instance)
             public string SessionId { get; }
 
             public Task<TResult> InvokeAsync<TResult>(string command, object input = null,
-                int timeoutSeconds = 0, CancellationToken cancellationToken = default) =>
-                Instance.InvokeAsync<TResult>(command, input, timeoutSeconds, cancellationToken, SessionId);
+                int timeoutSeconds = 0, CancellationToken cancellationToken = default,
+                IDictionary<string, string> properties = null) =>
+                Instance.InvokeAsync<TResult>(command, input, timeoutSeconds, cancellationToken,
+                    SessionId, properties);
 
             public ValueTask DisposeAsync() => new(pool.ReturnAsync(Instance, SessionId));
         }

SW.Serverless/Resident/IResidentAdapterHost.cs

@@ -17,7 +17,8 @@ public interface IAdapterLease : IAsyncDisposable
         /// disposal then clears through IResettable.
         /// </summary>
         Task<TResult> InvokeAsync<TResult>(string command, object input = null,
-            int timeoutSeconds = 0, CancellationToken cancellationToken = default);
+            int timeoutSeconds = 0, CancellationToken cancellationToken = default,
+            IDictionary<string, string> properties = null);
     }
 
     /// <summary>

SW.Serverless/Resident/ResidentAdapterInstance.cs

@@ -336,12 +336,18 @@ void WriteLog(LogEntry entry)
 
         void Send(HostFrame frame) => outbound.Writer.TryWrite(frame);
 
+        /// <summary>
+        /// <paramref name="properties"/> is configuration for THIS call, on top of the startup
+        /// values the process holds. An exclusive instance is shared by every caller pointed at it,
+        /// so anything that varies per caller has to travel with the call rather than with the
+        /// process.
+        /// </summary>
         public async Task<TResult> InvokeAsync<TResult>(string command, object input = null,
             int timeoutSeconds = 0, CancellationToken cancellationToken = default,
-            string sessionId = null)
+            string sessionId = null, IDictionary<string, string> properties = null)
         {
             var bytes = await InvokeAsync(command, Serialize(input), timeoutSeconds,
-                cancellationToken, sessionId);
+                cancellationToken, sessionId, properties);
             if (bytes == null || bytes.Length == 0) return default;
             if (typeof(TResult) == typeof(byte[])) return (TResult)(object)bytes;
             var text = System.Text.Encoding.UTF8.GetString(bytes);
@@ -351,7 +357,7 @@ public async Task<TResult> InvokeAsync<TResult>(string command, object input = n
 
         public async Task<byte[]> InvokeAsync(string command, byte[] payload = null,
             int timeoutSeconds = 0, CancellationToken cancellationToken = default,
-            string sessionId = null)
+            string sessionId = null, IDictionary<string, string> properties = null)
         {
             if (State != InstanceState.Ready)
                 throw new InvalidOperationException($"Adapter {AdapterId} is {State}, not Ready.");
@@ -380,17 +386,23 @@ public async Task<byte[]> InvokeAsync(string command, byte[] payload = null,
                 }
             }, null, timeout, Timeout.InfiniteTimeSpan);
 
+            var invoke = new Invoke
+            {
+                Command = command,
+                Payload = payload == null ? ByteString.Empty : ByteString.CopyFrom(payload),
+                TimeoutSeconds = (int)timeout.TotalSeconds,
+                SessionId = sessionId ?? ""
+            };
+
+            if (properties != null)
+                foreach (var kv in properties)
+                    invoke.Properties[kv.Key] = kv.Value ?? "";
+
             Send(new HostFrame
             {
                 Id = id,
                 Traceparent = Activity.Current?.Id ?? "",
-                Invoke = new Invoke
-                {
-                    Command = command,
-                    Payload = payload == null ? ByteString.Empty : ByteString.CopyFrom(payload),
-                    TimeoutSeconds = (int)timeout.TotalSeconds,
-                    SessionId = sessionId ?? ""
-                }
+                Invoke = invoke
             });
 
             using (cancellationToken.Register(() =>

docs/README.md

@@ -44,7 +44,8 @@ You get, without asking for it:
   `AdapterValues:` so it can never shadow them.
 * **`IAdapterContext`** for pushing events and metrics, injectable anywhere — and for reading and
   writing small durable state the host holds on the adapter's behalf (`GetStateAsync` /
-  `SetStateAsync`), which is where a polling receiver keeps its cursor.
+  `SetStateAsync`), which is where a polling receiver keeps its cursor, and for per-call
+  configuration on a shared instance (`ValueOf` / `InvocationValues`).
 * **`AdapterSession.Id`** — ambient per-invocation identity, and the boundary a pooled adapter
   needs so state cannot leak between checkouts.
 

docs/resident-adapters-design.md

@@ -1609,6 +1609,39 @@ Fixed by keying on `AdapterSpec.PoolKey` when set, and otherwise on the adapter
 the startup values, so identical configuration shares warm processes and differing configuration
 cannot. Hashed rather than concatenated because the key reaches logs and diagnostics.
 
+### 14.11 A shared instance needs per-call configuration
+
+§14.5 gave the exclusive resident one instance per key, and §4 assumed that instance had one
+caller. It does not. In Bitween a relational data source is one process holding one connection
+pool, and *every* subscription bound to that data source runs through it — each with its own
+settings: which statement to run, which operation it is, which tenant this is.
+
+Startup values cannot carry any of that. They are handed over once, in `Ready`, and they belong to
+the process — which belongs to all of those callers at once. So the invocation grows a
+`map<string, string> properties`, alongside the `session_id` that exists for the same underlying
+reason: a shared instance has to be told whose call this is.
+
+On the adapter side that surfaces as two members on `IAdapterContext`:
+
+```csharp
+IReadOnlyDictionary<string, string> InvocationValues { get; }  // this call's, empty outside one
+string ValueOf(string name);                                   // invocation first, then startup
+```
+
+`ValueOf` is the one to reach for. A per-call setting overrides the process default, and an adapter
+whose callers send no properties behaves exactly as it did before any of this existed — which is
+what keeps the existing fleet working.
+
+**It is an `AsyncLocal`, and that is the whole subtlety.** Several commands run on one instance at
+the same time; multiplexing is the point of the stream. A field would have the last caller in
+overwrite everyone else's configuration mid-flight, and the failure would be intermittent,
+load-dependent and near-impossible to reproduce — one subscription silently running another's
+statement. It is set on the invoking flow before the handler is called, so it survives the
+handler's own awaits and cannot leak sideways.
+
+Without this the shared-instance shape is only usable by callers that all want identical behaviour,
+which for a database connection is nobody.
+
 ---
 
 ## 15. What "gRPC over UDS / named pipe" actually means