From a052ae5b2b2400a2b718508ec0230e29151507bf Mon Sep 17 00:00:00 2001 From: eanzhao Date: Fri, 7 Aug 2026 04:49:48 +0800 Subject: [PATCH] Make cache child dispatch attributable and stale-safe MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 案例 20（supplier_risk_tier_aggregation）在生产出现一次 8/10 committed failed 且 lastError 为空，收敛到 cache miss 分支派发的动态 child 与 parent 都未完成。 根因未能从现有证据定位——失败 run 的标识未保留，而 cache 与终态判定路径上 的关键判定点没有 Information 级日志，107KB 生产日志里查不到任何 cache 记录。 本提交不猜测根因，只补齐让下次可归因的三处，并修一个已证实的派发缺陷： - CacheModule 的 StepCompletedEvent 分支此前完全没有日志，两处 early return 静默丢弃孤儿 child 完成，waiter 永久失联。现在按 child step marker 区分 「不是 cache child」与「映射丢失」，后者记 Warning；child 完成释放 waiter 时记 Information，与既有 HIT/MISS/PENDING 日志对称。 - 合成的 parent StepCompletedEvent 此前不带 ExecutionId，导致 WorkflowExecutionKernel 的 stale-execution 校验被跳过，超期派发的完成事件 会静默推进 run。现在把 kernel 分配给 parent 的 ExecutionId 存进 waiter 并原样回填。 - child StepRequestEvent 此前不转发 ExternalInvocation。kernel 明确要求派发 子步骤的原语复制该 call site，否则 child_step_type 为 tool_call / connector_call 时子步骤会丢失准入身份。foreach/while 已经这样做，cache 没有。 CacheWaiterState 新增 execution_id 字段（additive，旧状态默认空）。 未包含（另开 issue）：cache 尚不支持 sub_param_ 透传，因此需要参数的 child_step_type 仍会静默降级；这属于新增作者面，不在本次可观测性/正确性范围。 验证： - dotnet build src/workflow/Aevatar.Workflow.Core --nologo → 0 error - dotnet test test/Aevatar.Integration.Tests --nologo → 487 passed / 4 skipped （含新增 2 条：call site 转发 + ExecutionId 回填、映射丢失不误完成 parent） - dotnet test test/Aevatar.Workflow.Core.Tests --nologo → 946 passed - bash tools/ci/architecture_guards.sh → passed Co-Authored-By: Claude Opus 5 --- .../Modules/CacheModule.cs | 56 +++++++++++++- .../workflow_state.proto | 4 + .../WorkflowCoreModuleBehaviorTests.cs | 73 +++++++++++++++++++ 3 files changed, 129 insertions(+), 4 deletions(-) diff --git a/src/workflow/Aevatar.Workflow.Core/Modules/CacheModule.cs b/src/workflow/Aevatar.Workflow.Core/Modules/CacheModule.cs index c64db52286..d53b1bea57 100644 --- a/src/workflow/Aevatar.Workflow.Core/Modules/CacheModule.cs +++ b/src/workflow/Aevatar.Workflow.Core/Modules/CacheModule.cs @@ -15,6 +15,9 @@ public sealed class CacheModule : IEventModule { private const string ModuleStateKey = "cache"; + /// Marker embedded in dispatched child step ids so orphaned completions stay attributable. + 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(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, diff --git a/src/workflow/Aevatar.Workflow.Core/workflow_state.proto b/src/workflow/Aevatar.Workflow.Core/workflow_state.proto index fb5d9485d3..0ae27c4f73 100644 --- a/src/workflow/Aevatar.Workflow.Core/workflow_state.proto +++ b/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 diff --git a/test/Aevatar.Integration.Tests/WorkflowCoreModuleBehaviorTests.cs b/test/Aevatar.Integration.Tests/WorkflowCoreModuleBehaviorTests.cs index 6fd6f668d6..88c241d55b 100644 --- a/test/Aevatar.Integration.Tests/WorkflowCoreModuleBehaviorTests.cs +++ b/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().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().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() {