Make cache child dispatch attributable and stale-safe

src/workflow/Aevatar.Workflow.Core/Modules/CacheModule.cs

@@ -15,6 +15,9 @@ public sealed class CacheModule : IEventModule<IWorkflowExecutionContext>
 {
     private const string ModuleStateKey = "cache";
 
+    /// <summary>Marker embedded in dispatched child step ids so orphaned completions stay attributable.</summary>
+    private const string ChildStepMarker = "_cached_";
+
     public string Name => "cache";
     public int Priority => 3;
 
@@ -41,7 +44,12 @@ public async Task HandleAsync(EventEnvelope envelope, IWorkflowExecutionContext
             var cacheKey = request.Parameters.GetValueOrDefault("cache_key", request.Input ?? "");
             var ttlSeconds = int.TryParse(request.Parameters.GetValueOrDefault("ttl_seconds", "3600"), out var t) ? t : 3600;
             ttlSeconds = Math.Clamp(ttlSeconds, 1, 86_400);
-            var waiter = new CacheWaiterState { ParentStepId = request.StepId, RunId = runId };
+            var waiter = new CacheWaiterState
+            {
+                ParentStepId = request.StepId,
+                RunId = runId,
+                ExecutionId = request.ExecutionId ?? string.Empty,
+            };
 
             if (state.CacheEntries.TryGetValue(cacheKey, out var existingCache) &&
                 WorkflowTimestampCodec.ToDateTimeOffset(existingCache.ExpiresAt) <= now)
@@ -82,7 +90,7 @@ public async Task HandleAsync(EventEnvelope envelope, IWorkflowExecutionContext
             var childType = WorkflowPrimitiveCatalog.ToCanonicalType(
                 request.Parameters.GetValueOrDefault("child_step_type", "llm_call"));
             var childRole = request.Parameters.GetValueOrDefault("child_target_role", request.TargetRole);
-            var childStepId = $"{request.StepId}_cached_{Guid.NewGuid():N}";
+            var childStepId = $"{request.StepId}{ChildStepMarker}{Guid.NewGuid():N}";
 
             var pendingCall = new PendingCacheCallState
             {
@@ -93,14 +101,21 @@ public async Task HandleAsync(EventEnvelope envelope, IWorkflowExecutionContext
             state.ChildStepToCacheKey[BuildChildKey(runId, childStepId)] = cacheKey;
             await SaveStateAsync(state, ctx, ct);
 
-            await ctx.PublishAsync(new StepRequestEvent
+            var childRequest = new StepRequestEvent
             {
                 StepId = childStepId,
                 StepType = childType,
                 RunId = runId,
                 Input = request.Input ?? "",
                 TargetRole = childRole ?? "",
-            }, TopologyAudience.Self, ct);
+            };
+            // The kernel synthesizes the sub-step call site on the cache step itself and expects
+            // primitives that dispatch sub-steps to copy it onto every child they publish; without
+            // it an external child (tool_call/connector_call) loses its admission identity.
+            if (request.ExternalInvocation != null)
+                childRequest.ExternalInvocation = request.ExternalInvocation.Clone();
+
+            await ctx.PublishAsync(childRequest, TopologyAudience.Self, ct);
         }
         else if (payload.Is(StepCompletedEvent.Descriptor))
         {
@@ -109,9 +124,31 @@ await ctx.PublishAsync(new StepRequestEvent
             var state = WorkflowExecutionStateAccess.Load<CacheModuleState>(ctx, ModuleStateKey);
             var childKey = BuildChildKey(runId, evt.StepId);
             if (!state.ChildStepToCacheKey.Remove(childKey, out var cacheKey))
+            {
+                // Not a cache child (the vast majority of completions) — stay quiet. Only warn when
+                // the step id carries the cache child marker, which means the parent↔child mapping
+                // was lost and every waiter on it is now stranded with no other recovery path.
+                if (evt.StepId.Contains(ChildStepMarker, StringComparison.Ordinal))
+                {
+                    ctx.Logger.LogWarning(
+                        "Cache {StepId}: orphan child completion, no parent mapping. run={RunId} success={Success}",
+                        evt.StepId,
+                        runId,
+                        evt.Success);
+                }
+
                 return;
+            }
+
             if (!state.PendingByCacheKey.Remove(cacheKey, out var pending))
+            {
+                ctx.Logger.LogWarning(
+                    "Cache {StepId}: child mapped to key={Key} but no pending call remains; waiters stranded. run={RunId}",
+                    evt.StepId,
+                    ShortenKey(cacheKey),
+                    runId);
                 return;
+            }
 
             if (evt.Success)
             {
@@ -123,12 +160,23 @@ await ctx.PublishAsync(new StepRequestEvent
             }
             await SaveStateAsync(state, ctx, ct);
 
+            ctx.Logger.LogInformation(
+                "Cache {StepId}: child completed key={Key} success={Success}, releasing waiters={Waiters}",
+                evt.StepId,
+                ShortenKey(cacheKey),
+                evt.Success,
+                pending.Waiters.Count);
+
             foreach (var waiter in pending.Waiters)
             {
                 var completed = new StepCompletedEvent
                 {
                     StepId = waiter.ParentStepId,
                     RunId = waiter.RunId,
+                    // Carry the dispatch identity the kernel assigned to the parent step so a
+                    // completion synthesized for a superseded dispatch is rejected instead of
+                    // silently advancing the run.
+                    ExecutionId = waiter.ExecutionId ?? string.Empty,
                     Success = evt.Success,
                     Output = evt.Output,
                     Error = evt.Error,

src/workflow/Aevatar.Workflow.Core/workflow_state.proto

@@ -758,6 +758,10 @@ message CacheWaiterState
 {
   string run_id = 1;
   string parent_step_id = 2;
+  // Execution id assigned by the kernel when it dispatched the parent cache step.
+  // Echoed back on the synthesized parent completion so the kernel's stale-execution
+  // guard can reject completions belonging to a superseded dispatch.
+  string execution_id = 3;
 }
 
 message PendingCacheCallState

test/Aevatar.Integration.Tests/WorkflowCoreModuleBehaviorTests.cs

@@ -742,6 +742,79 @@ await module.HandleAsync(
             hitCompletion.Annotations.Should().ContainKey("cache.key");
         }
 
+        [Fact]
+        public async Task CacheModule_OnMiss_ShouldForwardCallSiteAndEchoParentExecutionIdOnCompletion()
+        {
+            var module = new CacheModule();
+            var ctx = CreateContext();
+
+            await module.HandleAsync(
+                Envelope(new StepRequestEvent
+                {
+                    StepId = "cache-parent-call-site",
+                    StepType = "cache",
+                    RunId = "run-cache-call-site",
+                    Input = "input-call-site",
+                    ExecutionId = "exec-parent-1",
+                    ExternalInvocation = new ExternalToolInvocationSpec
+                    {
+                        CallSiteId = "call-site-1",
+                        ToolName = "nyxid_proxy",
+                    },
+                    Parameters =
+                    {
+                        ["cache_key"] = "key-call-site",
+                        ["child_step_type"] = "tool_call",
+                    },
+                }),
+                ctx,
+                CancellationToken.None);
+
+            var childRequest = ctx.Published.Select(x => x.evt).OfType<StepRequestEvent>().Single();
+            childRequest.StepType.Should().Be("tool_call");
+            childRequest.ExternalInvocation.Should().NotBeNull();
+            childRequest.ExternalInvocation!.CallSiteId.Should().Be("call-site-1");
+            childRequest.ExternalInvocation.ToolName.Should().Be("nyxid_proxy");
+
+            ctx.Published.Clear();
+
+            await module.HandleAsync(
+                Envelope(new StepCompletedEvent
+                {
+                    StepId = childRequest.StepId,
+                    RunId = "run-cache-call-site",
+                    Success = true,
+                    Output = "call-site-output",
+                }),
+                ctx,
+                CancellationToken.None);
+
+            var completion = ctx.Published.Select(x => x.evt).OfType<StepCompletedEvent>().Single();
+            completion.StepId.Should().Be("cache-parent-call-site");
+            completion.ExecutionId.Should().Be("exec-parent-1");
+            completion.Output.Should().Be("call-site-output");
+        }
+
+        [Fact]
+        public async Task CacheModule_WhenChildMappingIsLost_ShouldNotCompleteAnyParent()
+        {
+            var module = new CacheModule();
+            var ctx = CreateContext();
+
+            await module.HandleAsync(
+                Envelope(new StepCompletedEvent
+                {
+                    StepId = "cache-parent-orphan_cached_deadbeef",
+                    RunId = "run-cache-orphan",
+                    Success = true,
+                    Output = "orphan-output",
+                }),
+                ctx,
+                CancellationToken.None);
+
+            ctx.Published.Should().BeEmpty();
+        }
+
         [Fact]
         public async Task CacheModule_WhenSecondCallerJoinsPending_ShouldFanOutCompletionAndNotCacheFailures()
         {