Issue #3252: [Backend] Workflow run durable lineage contract

docs/canon/workflow-runtime.md

@@ -67,6 +67,7 @@ owner: eanzhao
    - 通过依赖推导（`IWorkflowModuleDependencyExpander`）确定所需模块，经 `WorkflowModuleFactory` 创建并安装
    - 收到 `ChatRequestEvent` envelope 后发布 `StartWorkflowEvent`
    - fork/resume-from-step seed 只走 request-level `WorkflowChatRequestEvent.fork_seed -> StartWorkflowEvent.fork_seed`；run bind 只表达 definition/run binding，不携带 seed。
+   - run lineage 是 `WorkflowRunGAgent` owned committed fact，不从 route、actor id、graph/topology、workflow name 或 ID 前缀推断。`WorkflowRunLineage` 分离 retry/fork 与 `workflow_call` parent/child 关系，并始终使用可路由 public `runId`；actor address 只作为可选寻址信息保留。legacy 或未携带 lineage 的 run 必须显式返回 unavailable/legacy-unavailable。
    - 由 `WorkflowExecutionKernel` 推进 `StepRequestEvent -> StepCompletedEvent -> WorkflowCompletedEvent`
 
 ```

src/workflow/Aevatar.Workflow.Abstractions/workflow_execution_messages.proto

@@ -270,6 +270,7 @@ message WorkflowRunForkSeed
   map<string, string> variables = 3;
   int32 attempt = 4;
   WorkflowStepIdempotencyState start_step_idempotency = 5;
+  string original_run_id = 6;
 }
 
 message WorkflowStepIdempotencyState
@@ -280,6 +281,60 @@ message WorkflowStepIdempotencyState
   string idempotency_key = 4;
 }
 
+enum WorkflowRunLineageAvailability
+{
+  WORKFLOW_RUN_LINEAGE_AVAILABILITY_UNSPECIFIED = 0;
+  WORKFLOW_RUN_LINEAGE_AVAILABILITY_AVAILABLE = 1;
+  WORKFLOW_RUN_LINEAGE_AVAILABILITY_UNAVAILABLE = 2;
+  WORKFLOW_RUN_LINEAGE_AVAILABILITY_LEGACY_UNAVAILABLE = 3;
+}
+
+enum WorkflowRunLineageRelationKind
+{
+  WORKFLOW_RUN_LINEAGE_RELATION_KIND_UNSPECIFIED = 0;
+  WORKFLOW_RUN_LINEAGE_RELATION_KIND_RETRY_FORK = 1;
+  WORKFLOW_RUN_LINEAGE_RELATION_KIND_SUB_WORKFLOW = 2;
+}
+
+message WorkflowRunLineageRunRef
+{
+  string run_id = 1;
+  string actor_id = 2;
+  string relationship_id = 3;
+  string step_id = 4;
+  int32 attempt = 5;
+  WorkflowRunLineageRelationKind relation_kind = 6;
+}
+
+message WorkflowRunRetryForkLineage
+{
+  WorkflowRunLineageAvailability availability = 1;
+  string source_run_id = 2;
+  string original_run_id = 3;
+  int32 attempt = 4;
+  string start_at_step_id = 5;
+  repeated WorkflowRunLineageRunRef child_runs = 6;
+}
+
+message WorkflowRunSubWorkflowLineage
+{
+  WorkflowRunLineageAvailability availability = 1;
+  string parent_run_id = 2;
+  string parent_actor_id = 3;
+  string parent_step_id = 4;
+  string root_run_id = 5;
+  int32 depth = 6;
+  repeated WorkflowRunLineageRunRef child_runs = 7;
+}
+
+message WorkflowRunLineage
+{
+  WorkflowRunLineageAvailability availability = 1;
+  WorkflowRunRetryForkLineage retry_fork = 2;
+  WorkflowRunSubWorkflowLineage sub_workflow = 3;
+  string unavailable_reason = 4;
+}
+
 message WorkflowRunForkRequestedEvent
 {
   string source_run_id = 1;
@@ -288,6 +343,17 @@ message WorkflowRunForkRequestedEvent
   string scope_id = 4;
 }
 
+message WorkflowRunLineageRecordedEvent
+{
+  string source_run_id = 1;
+  string child_run_id = 2;
+  string child_actor_id = 3;
+  string start_at_step_id = 4;
+  int32 attempt = 5;
+  WorkflowRunLineageRelationKind relation_kind = 6;
+  string original_run_id = 7;
+}
+
 message StartWorkflowEvent
 {
   string workflow_name = 1;
@@ -324,6 +390,7 @@ message BindWorkflowRunDefinitionEvent
   string revision_id = 12;
   // Authoritative committed definition version/watermark used by this run. Zero means unavailable.
   int64 definition_version = 13;
+  WorkflowRunLineage initial_lineage = 14;
 }
 
 // Idempotent provisioning command for a stable Run actor identity. The first
@@ -380,6 +447,7 @@ message WorkflowRunExecutionStartedEvent
   string workflow_command_id = 11;
   string workflow_correlation_id = 12;
   string current_turn_id = 13;
+  WorkflowRunLineage lineage = 14;
 }
 message WorkflowRunExecutionContextDelta
 {

src/workflow/Aevatar.Workflow.Application.Abstractions/Observatory/IWorkflowRunObservatoryQueryService.cs

@@ -1,3 +1,4 @@
+using Aevatar.Workflow.Abstractions;
 using Aevatar.Workflow.Application.Abstractions.Queries;
 
 namespace Aevatar.Workflow.Application.Abstractions.Observatory;
@@ -122,6 +123,8 @@ public sealed class WorkflowActivityRunFeedRow
     public long StateVersion { get; init; }
 
     public WorkflowRunRecoveryCapability RecoveryCapability { get; init; } = new();
+
+    public WorkflowRunLineage Lineage { get; init; } = new();
 }
 
 public sealed class WorkflowActivityRunInitiatorSummary
@@ -196,6 +199,8 @@ public sealed class ObservatoryRunSummary
     // 06-23-observatory-run-coverage-filter: canonical run origin/type (draft | member-invoke | ...),
     // empty for legacy/unstamped runs. Drives the run-type filter + badge.
     public string RunOrigin { get; init; } = string.Empty;
+
+    public WorkflowRunLineage Lineage { get; init; } = new();
 }
 
 public sealed class ObservatoryRunDetail
@@ -227,6 +232,8 @@ public sealed class ObservatoryRunDetail
     public ObservatoryUsageTotals UsageTotals { get; init; } = new();
 
     public WorkflowRunRecoveryCapability RecoveryCapability { get; init; } = new();
+
+    public WorkflowRunLineage Lineage { get; init; } = new();
 }
 
 public sealed class ObservatoryRunDiagnostic

src/workflow/Aevatar.Workflow.Application.Abstractions/Queries/workflow_projection_query_models.proto

@@ -89,6 +89,7 @@ message WorkflowActorSnapshot {
   WorkflowRunActivityWaitingSnapshot activity_waiting = 31;
   string run_id = 32;
   WorkflowRunRecoveryCapability recovery_capability = 33;
+  aevatar.workflow.WorkflowRunLineage lineage = 34;
 }
 
 message WorkflowRunActivityInitiatorSnapshot {

src/workflow/Aevatar.Workflow.Application.Abstractions/RunForks/WorkflowForkRunModels.cs

@@ -122,4 +122,5 @@ public sealed record WorkflowForkRunAcceptedReceipt(
     string CommandId,
     string CorrelationId,
     DateTimeOffset AckedAt,
-    string NewRunId = "");
+    string NewRunId = "",
+    string OriginalRunId = "");

src/workflow/Aevatar.Workflow.Application.Abstractions/Runs/WorkflowChatRunModels.cs

@@ -147,7 +147,8 @@ public sealed record WorkflowChatRunForkSeed(
     string StartAtStepId,
     IReadOnlyDictionary<string, string> Variables,
     int Attempt = 0,
-    WorkflowStepIdempotencyView? StartStepIdempotency = null);
+    WorkflowStepIdempotencyView? StartStepIdempotency = null,
+    string OriginalRunId = "");
 
 public enum WorkflowChatSourceKind
 {

src/workflow/Aevatar.Workflow.Application.Abstractions/Runs/WorkflowRunPorts.cs

@@ -165,7 +165,8 @@ public sealed record WorkflowRunForkSeedView(
     IReadOnlyDictionary<string, WorkflowStepIdempotencyView>? IdempotencyByStepId = null,
     WorkflowCapabilityAdmissionPlan? CapabilityAdmissionPlan = null,
     string RevisionId = "",
-    long DefinitionVersion = 0)
+    long DefinitionVersion = 0,
+    string OriginalRunId = "")
 {
     public WorkflowRunForkSeedView()
         : this(
@@ -182,7 +183,8 @@ public WorkflowRunForkSeedView()
             new Dictionary<string, WorkflowStepIdempotencyView>(StringComparer.Ordinal),
             null,
             string.Empty,
-            0)
+            0,
+            string.Empty)
     {
     }
 }

src/workflow/Aevatar.Workflow.Application/Observatory/WorkflowRunObservatoryQueryService.cs

@@ -231,6 +231,7 @@ private async Task<ObservatoryRunDetail> BuildRunDetailAsync(
                 Diagnostics = BuildDiagnostics(snapshot, report: null, steps: [], viewEvents: []),
                 UsageTotals = new ObservatoryUsageTotals(),
                 RecoveryCapability = CloneRecoveryCapability(snapshot),
+                Lineage = CloneLineage(snapshot),
             };
         }
 
@@ -266,6 +267,7 @@ private async Task<ObservatoryRunDetail> BuildRunDetailAsync(
             Statistics = ToStatistics(report.Summary),
             UsageTotals = WorkflowRunObservatoryTimelineMapper.ToUsageTotals(report.Usage),
             RecoveryCapability = CloneRecoveryCapability(snapshot),
+            Lineage = CloneLineage(snapshot),
         };
     }
 
@@ -378,6 +380,7 @@ private static ObservatoryRunSummary ToRunSummary(WorkflowActorSnapshot snapshot
             StateVersion = snapshot.StateVersion,
             ScopeId = snapshot.ScopeId,
             RunOrigin = snapshot.RunOrigin,
+            Lineage = CloneLineage(snapshot),
         };
     }
 
@@ -408,12 +411,28 @@ private static WorkflowActivityRunFeedRow ToActivityRunFeedRow(WorkflowActorSnap
             DurationMs = completedAtUtc == null || !snapshot.HasDurationMs ? null : snapshot.DurationMs,
             StateVersion = snapshot.StateVersion,
             RecoveryCapability = CloneRecoveryCapability(snapshot),
+            Lineage = CloneLineage(snapshot),
         };
     }
 
     private static WorkflowRunRecoveryCapability CloneRecoveryCapability(WorkflowActorSnapshot snapshot) =>
         snapshot.RecoveryCapability?.Clone() ?? new WorkflowRunRecoveryCapability();
 
+    private static Aevatar.Workflow.Abstractions.WorkflowRunLineage CloneLineage(WorkflowActorSnapshot snapshot) =>
+        snapshot.Lineage?.Clone() ?? new Aevatar.Workflow.Abstractions.WorkflowRunLineage
+        {
+            Availability = Aevatar.Workflow.Abstractions.WorkflowRunLineageAvailability.LegacyUnavailable,
+            UnavailableReason = "Run lineage is unavailable for this legacy run.",
+            RetryFork = new Aevatar.Workflow.Abstractions.WorkflowRunRetryForkLineage
+            {
+                Availability = Aevatar.Workflow.Abstractions.WorkflowRunLineageAvailability.LegacyUnavailable,
+            },
+            SubWorkflow = new Aevatar.Workflow.Abstractions.WorkflowRunSubWorkflowLineage
+            {
+                Availability = Aevatar.Workflow.Abstractions.WorkflowRunLineageAvailability.LegacyUnavailable,
+            },
+        };
+
     private static WorkflowActivityRunInitiatorSummary ToActivityInitiatorSummary(
         WorkflowRunActivityInitiatorSnapshot? source) =>
         source == null

src/workflow/Aevatar.Workflow.Application/RunForks/WorkflowForkRunAcceptedReceiptFactory.cs

@@ -21,6 +21,7 @@ public WorkflowForkRunAcceptedReceipt Create(
             context.CommandId,
             context.CorrelationId,
             DateTimeOffset.UtcNow,
-            target.RunId);
+            target.RunId,
+            target.OriginalRunId);
     }
 }

src/workflow/Aevatar.Workflow.Application/RunForks/WorkflowForkRunCommandTarget.cs

@@ -10,6 +10,7 @@ internal sealed class WorkflowForkRunCommandTarget : ICommandDispatchTarget, ICo
 
     public WorkflowForkRunCommandTarget(
         string sourceRunId,
+        string originalRunId,
         string startAtStepId,
         string actorId,
         string runId,
@@ -19,6 +20,7 @@ public WorkflowForkRunCommandTarget(
         IWorkflowRunProvisioningPort runProvisioningPort)
     {
         SourceRunId = Normalize(sourceRunId);
+        OriginalRunId = Normalize(originalRunId);
         StartAtStepId = Normalize(startAtStepId);
         RunId = Normalize(runId);
         PreparedRequest = preparedRequest ?? throw new ArgumentNullException(nameof(preparedRequest));
@@ -31,6 +33,8 @@ public WorkflowForkRunCommandTarget(
 
     public string SourceRunId { get; }
 
+    public string OriginalRunId { get; }
+
     public string StartAtStepId { get; }
 
     public string RunId { get; }

src/workflow/Aevatar.Workflow.Application/RunForks/WorkflowForkRunCommandTargetResolver.cs

@@ -105,6 +105,7 @@ public async Task<CommandTargetResolution<WorkflowForkRunCommandTarget, Workflow
         var source = WorkflowChatSource.DefinitionActor(creationReceipt.ActorId, validation.WorkflowName);
         var target = new WorkflowForkRunCommandTarget(
             sourceRunId,
+            ResolveOriginalRunId(seedView, sourceRunId),
             startAtStepId,
             creationReceipt.ActorId,
             creationReceipt.RunId,
@@ -196,7 +197,8 @@ private static WorkflowChatRunRequest BuildChatRunRequest(
                 startAtStepId,
                 variables,
                 Math.Max(0, command.Attempt),
-                ResolveStartStepIdempotency(seedView, startAtStepId)),
+                ResolveStartStepIdempotency(seedView, startAtStepId),
+                ResolveOriginalRunId(seedView, sourceRunId)),
             TargetSeed: new WorkflowRunTargetSeed(
                 actorId,
                 workflowName,
@@ -272,6 +274,14 @@ private static string ResolveResumeInput(
             : null;
     }
 
+    private static string ResolveOriginalRunId(
+        WorkflowRunForkSeedView seedView,
+        string sourceRunId)
+    {
+        var originalRunId = Normalize(seedView.OriginalRunId);
+        return string.IsNullOrWhiteSpace(originalRunId) ? sourceRunId : originalRunId;
+    }
+
     private static bool IsTerminal(string status) =>
         !string.IsNullOrWhiteSpace(status) && TerminalStatuses.Contains(status.Trim());