Skip to content

Issue #3251: [Backend] Workflow run recovery capability and receipt contract #3293

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
louis4li wants to merge 2 commits into
base: feat/2026-08-07_issue-3250
Choose a base branch
from
Open

Issue #3251: [Backend] Workflow run recovery capability and receipt contract #3293

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
df0cc74
Issue #3251: Workflow run recovery capability and receipt contract
louis4li Aug 7, 2026
33e1335
Fix review round 1 on PR #3293
louis4li Aug 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions src/workflow/Aevatar.Workflow.Abstractions/workflow_execution_messages.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -320,6 +320,10 @@ message BindWorkflowRunDefinitionEvent
// Authoritative draft/workspace workflow identity carried from the definition binding.
// Empty means the run is legacy or not backed by a draft workflow identity.
string workflow_id = 11;
// Exact definition revision used by this run. Empty for legacy or ad-hoc runs.
string revision_id = 12;
// Authoritative committed definition version/watermark used by this run. Zero means unavailable.
int64 definition_version = 13;
}

// Idempotent provisioning command for a stable Run actor identity. The first
Expand Down Expand Up @@ -416,13 +420,21 @@ message WorkflowStoppedEvent {
string reason = 3;
google.protobuf.Timestamp completed_at_utc = 4;
}

enum WorkflowRecoveryFailureKind {
WORKFLOW_RECOVERY_FAILURE_KIND_UNSPECIFIED = 0;
WORKFLOW_RECOVERY_FAILURE_KIND_AUTHORIZATION_FAILURE = 1;
WORKFLOW_RECOVERY_FAILURE_KIND_CONFIGURATION_FAILURE = 2;
}

message WorkflowCompletedEvent {
string workflow_name = 1;
bool success = 2;
string output = 3;
string error = 4;
string run_id = 5;
google.protobuf.Timestamp completed_at_utc = 6;
WorkflowRecoveryFailureKind recovery_failure_kind = 7;
}

message WorkflowRunTerminalTimingRecordedEvent {
Expand Down Expand Up @@ -642,6 +654,7 @@ message StepCompletedEvent {
VoteAgreementDecision vote_agreement_decision = 14;
WorkflowStepFailureOutcome failure_outcome = 15;
WorkflowFileItemResultSet file_item_results = 16;
WorkflowRecoveryFailureKind recovery_failure_kind = 17;
}
message SubWorkflowInvokeRequestedEvent
{
Expand All @@ -664,6 +677,7 @@ message WorkflowDefinitionSnapshot
map<string, string> inline_workflow_yamls = 4;
string scope_id = 5;
int32 definition_version = 6;
string revision_id = 7;
}
message SubWorkflowDefinitionResolveRequestedEvent
{
Expand Down Expand Up @@ -958,6 +972,7 @@ message WorkflowLlmInvocationCompletedEvent
WorkflowUsageMetrics usage = 9;
WorkflowManagedHandoffOutcome managed_handoff = 10;
WorkflowInteractiveAuthorizationRequirement authorization_requirement = 11;
WorkflowRecoveryFailureKind recovery_failure_kind = 12;
}

message WorkflowInteractiveAuthorizationRequirement
Expand Down
6 changes: 6 additions & 0 deletions ...atar.Workflow.Application.Abstractions/Observatory/IWorkflowRunObservatoryQueryService.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
using Aevatar.Workflow.Application.Abstractions.Queries;

namespace Aevatar.Workflow.Application.Abstractions.Observatory;

// 06-19-workflow-run-observatory (C2): read-only, scope-gated run viewer query port.
Expand Down Expand Up @@ -118,6 +120,8 @@ public sealed class WorkflowActivityRunFeedRow
public double? DurationMs { get; init; }

public long StateVersion { get; init; }

public WorkflowRunRecoveryCapability RecoveryCapability { get; init; } = new();
}

public sealed class WorkflowActivityRunInitiatorSummary
Expand Down Expand Up @@ -221,6 +225,8 @@ public sealed class ObservatoryRunDetail
public ObservatoryRunStatistics Statistics { get; init; } = new();

public ObservatoryUsageTotals UsageTotals { get; init; } = new();

public WorkflowRunRecoveryCapability RecoveryCapability { get; init; } = new();
}

public sealed class ObservatoryRunDiagnostic
Expand Down
47 changes: 47 additions & 0 deletions .../Aevatar.Workflow.Application.Abstractions/Queries/workflow_projection_query_models.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,52 @@ import "google/protobuf/wrappers.proto";
import "workflow_execution_messages.proto";
import "workflow_external_action.proto";

enum WorkflowRecoveryEligibility {
WORKFLOW_RECOVERY_ELIGIBILITY_UNSPECIFIED = 0;
WORKFLOW_RECOVERY_ELIGIBILITY_ELIGIBLE = 1;
WORKFLOW_RECOVERY_ELIGIBILITY_INELIGIBLE = 2;
WORKFLOW_RECOVERY_ELIGIBILITY_UNAVAILABLE = 3;
}

enum WorkflowRecoveryUnavailableReasonCode {
WORKFLOW_RECOVERY_UNAVAILABLE_REASON_CODE_UNSPECIFIED = 0;
WORKFLOW_RECOVERY_UNAVAILABLE_REASON_CODE_NONE = 1;
WORKFLOW_RECOVERY_UNAVAILABLE_REASON_CODE_SOURCE_RUN_NOT_TERMINAL = 2;
WORKFLOW_RECOVERY_UNAVAILABLE_REASON_CODE_MISSING_SOURCE_FACT = 3;
WORKFLOW_RECOVERY_UNAVAILABLE_REASON_CODE_AUTHORIZATION_FAILURE = 4;
WORKFLOW_RECOVERY_UNAVAILABLE_REASON_CODE_CONFIGURATION_FAILURE = 5;
WORKFLOW_RECOVERY_UNAVAILABLE_REASON_CODE_WORKFLOW_DEFINITION_UNAVAILABLE = 6;
WORKFLOW_RECOVERY_UNAVAILABLE_REASON_CODE_LEGACY_UNAVAILABLE = 7;
}

enum WorkflowRecoveryRecommendedAction {
WORKFLOW_RECOVERY_RECOMMENDED_ACTION_UNSPECIFIED = 0;
WORKFLOW_RECOVERY_RECOMMENDED_ACTION_RETRY = 1;
WORKFLOW_RECOVERY_RECOMMENDED_ACTION_RUN_AGAIN = 2;
WORKFLOW_RECOVERY_RECOMMENDED_ACTION_FIX_ACCESS = 3;
WORKFLOW_RECOVERY_RECOMMENDED_ACTION_CHANGE_CONFIGURATION = 4;
WORKFLOW_RECOVERY_RECOMMENDED_ACTION_EDIT_WORKFLOW = 5;
WORKFLOW_RECOVERY_RECOMMENDED_ACTION_EDIT_INPUT = 6;
WORKFLOW_RECOVERY_RECOMMENDED_ACTION_TECHNICAL_DETAILS = 7;
}

message WorkflowRecoveryActionCapability {
WorkflowRecoveryEligibility eligibility = 1;
WorkflowRecoveryUnavailableReasonCode unavailable_reason_code = 2;
string unavailable_reason = 3;
repeated WorkflowRecoveryRecommendedAction recommended_actions = 4;
string starting_step_id = 5;
bool reuses_prior_step_outputs = 6;
bool may_incur_model_or_tool_cost = 7;
}

message WorkflowRunRecoveryCapability {
WorkflowRecoveryActionCapability retry_failed_step = 1;
WorkflowRecoveryActionCapability run_again = 2;
string workflow_definition_revision_id = 3;
int64 workflow_definition_version = 4;
}

message WorkflowActorSnapshot {
string actor_id = 1;
string workflow_name = 2;
Expand Down Expand Up @@ -42,6 +88,7 @@ message WorkflowActorSnapshot {
WorkflowRunActivityFailureSnapshot activity_first_failure = 30;
WorkflowRunActivityWaitingSnapshot activity_waiting = 31;
string run_id = 32;
WorkflowRunRecoveryCapability recovery_capability = 33;
}

message WorkflowRunActivityInitiatorSnapshot {
Expand Down
3 changes: 2 additions & 1 deletion src/workflow/Aevatar.Workflow.Application.Abstractions/RunForks/WorkflowForkRunModels.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -121,4 +121,5 @@ public sealed record WorkflowForkRunAcceptedReceipt(
bool Accepted,
string CommandId,
string CorrelationId,
DateTimeOffset AckedAt);
DateTimeOffset AckedAt,
string NewRunId = "");
14 changes: 10 additions & 4 deletions src/workflow/Aevatar.Workflow.Application.Abstractions/Runs/WorkflowRunPorts.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,12 +87,14 @@ public sealed record WorkflowDefinitionBinding(
string SourceKind = "",
WorkflowCapabilityAdmissionPlan? CapabilityAdmissionPlan = null,
string WorkflowId = "",
string RevisionId = "");
string RevisionId = "",
long DefinitionVersion = 0);

public sealed record WorkflowRunCreationReceipt(
string ActorId,
string DefinitionActorId,
IReadOnlyList<string> CreatedActorIds);
IReadOnlyList<string> CreatedActorIds,
string RunId = "");

public sealed record WorkflowDefinitionProvisioningReceipt(
string ActorId,
Expand Down Expand Up @@ -161,7 +163,9 @@ public sealed record WorkflowRunForkSeedView(
string FinalError,
string ScopeId = "",
IReadOnlyDictionary<string, WorkflowStepIdempotencyView>? IdempotencyByStepId = null,
WorkflowCapabilityAdmissionPlan? CapabilityAdmissionPlan = null)
WorkflowCapabilityAdmissionPlan? CapabilityAdmissionPlan = null,
string RevisionId = "",
long DefinitionVersion = 0)
{
public WorkflowRunForkSeedView()
: this(
Expand All @@ -176,7 +180,9 @@ public WorkflowRunForkSeedView()
string.Empty,
string.Empty,
new Dictionary<string, WorkflowStepIdempotencyView>(StringComparer.Ordinal),
null)
null,
string.Empty,
0)
{
}
}
Expand Down
6 changes: 6 additions & 0 deletions src/workflow/Aevatar.Workflow.Application/Observatory/WorkflowRunObservatoryQueryService.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -230,6 +230,7 @@ private async Task<ObservatoryRunDetail> BuildRunDetailAsync(
Timeline = [],
Diagnostics = BuildDiagnostics(snapshot, report: null, steps: [], viewEvents: []),
UsageTotals = new ObservatoryUsageTotals(),
RecoveryCapability = CloneRecoveryCapability(snapshot),
};
}

Expand Down Expand Up @@ -264,6 +265,7 @@ private async Task<ObservatoryRunDetail> BuildRunDetailAsync(
Timeline = viewEvents,
Statistics = ToStatistics(report.Summary),
UsageTotals = WorkflowRunObservatoryTimelineMapper.ToUsageTotals(report.Usage),
RecoveryCapability = CloneRecoveryCapability(snapshot),
};
}

Expand Down Expand Up @@ -405,9 +407,13 @@ private static WorkflowActivityRunFeedRow ToActivityRunFeedRow(WorkflowActorSnap
// Read the optional snapshot field so completed-without-start stays unavailable.
DurationMs = completedAtUtc == null || !snapshot.HasDurationMs ? null : snapshot.DurationMs,
StateVersion = snapshot.StateVersion,
RecoveryCapability = CloneRecoveryCapability(snapshot),
};
}

private static WorkflowRunRecoveryCapability CloneRecoveryCapability(WorkflowActorSnapshot snapshot) =>
snapshot.RecoveryCapability?.Clone() ?? new WorkflowRunRecoveryCapability();

private static WorkflowActivityRunInitiatorSummary ToActivityInitiatorSummary(
WorkflowRunActivityInitiatorSnapshot? source) =>
source == null
Expand Down
3 changes: 2 additions & 1 deletion src/workflow/Aevatar.Workflow.Application/RunForks/WorkflowForkRunAcceptedReceiptFactory.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ public WorkflowForkRunAcceptedReceipt Create(
true,
context.CommandId,
context.CorrelationId,
DateTimeOffset.UtcNow);
DateTimeOffset.UtcNow,
target.RunId);
}
}
4 changes: 4 additions & 0 deletions src/workflow/Aevatar.Workflow.Application/RunForks/WorkflowForkRunCommandTarget.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,13 +12,15 @@ public WorkflowForkRunCommandTarget(
string sourceRunId,
string startAtStepId,
string actorId,
string runId,
string workflowName,
WorkflowChatRunRequest preparedRequest,
IReadOnlyList<string>? createdActorIds,
IWorkflowRunProvisioningPort runProvisioningPort)
{
SourceRunId = Normalize(sourceRunId);
StartAtStepId = Normalize(startAtStepId);
RunId = Normalize(runId);
PreparedRequest = preparedRequest ?? throw new ArgumentNullException(nameof(preparedRequest));
_innerTarget = new WorkflowRunAcceptedCommandTarget(
actorId,
Expand All @@ -31,6 +33,8 @@ public WorkflowForkRunCommandTarget(

public string StartAtStepId { get; }

public string RunId { get; }

public WorkflowChatRunRequest PreparedRequest { get; }

public string ActorId => _innerTarget.ActorId;
Expand Down
5 changes: 4 additions & 1 deletion src/workflow/Aevatar.Workflow.Application/RunForks/WorkflowForkRunCommandTargetResolver.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,9 @@ public async Task<CommandTargetResolution<WorkflowForkRunCommandTarget, Workflow
InlineWorkflowYamls: inlineWorkflowYamls,
ExpectedExecutionMode: seedView.ExpectedExecutionMode,
ScopeId: scopeId,
CapabilityAdmissionPlan: seedView.CapabilityAdmissionPlan?.Clone()),
CapabilityAdmissionPlan: seedView.CapabilityAdmissionPlan?.Clone(),
RevisionId: seedView.RevisionId,
DefinitionVersion: Math.Max(0, seedView.DefinitionVersion)),
Comment on lines +95 to +96

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Clear source revision when workflow YAML is overridden

When InlineYaml or InlineSubYamls changes the source definition, this still stamps the new run with the source run's RevisionId and DefinitionVersion. The resulting committed state and recovery capability therefore claim an exact revision that the run did not execute, corrupting provenance and potentially causing later recovery to resolve the wrong definition; preserve these fields only when the effective YAML bundle is unchanged, otherwise resolve or clear the revision identity.

AGENTS.md reference: AGENTS.md:L62-L63

Useful? React with 👍 / 👎.

ct).ConfigureAwait(false);
}
catch (Exception ex)
Expand All @@ -105,6 +107,7 @@ public async Task<CommandTargetResolution<WorkflowForkRunCommandTarget, Workflow
sourceRunId,
startAtStepId,
creationReceipt.ActorId,
creationReceipt.RunId,
validation.WorkflowName,
BuildChatRunRequest(
command,
Expand Down
6 changes: 4 additions & 2 deletions src/workflow/Aevatar.Workflow.Application/Runs/WorkflowRunActorResolver.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -296,7 +296,8 @@ private async Task<WorkflowActorResolutionResult> ResolveFromSourceActorAsync(
SourceKind: sourceBinding.SourceKind,
CapabilityAdmissionPlan: sourceBinding.CapabilityAdmissionPlan?.Clone(),
WorkflowId: sourceBinding.WorkflowId,
RevisionId: sourceBinding.RevisionId),
RevisionId: sourceBinding.RevisionId,
DefinitionVersion: Math.Max(0, sourceBinding.SourceVersion)),
wrapAsFallbackTrigger: true,
ct);

Expand Down Expand Up @@ -373,7 +374,8 @@ private async Task<WorkflowActorResolutionResult> ResolveFromResolvedDefinitionB
SourceKind: resolvedDefinitionBinding.SourceKind?.Trim() ?? string.Empty,
CapabilityAdmissionPlan: resolvedDefinitionBinding.CapabilityAdmissionPlan?.Clone(),
WorkflowId: resolvedDefinitionBinding.WorkflowId?.Trim() ?? string.Empty,
RevisionId: resolvedDefinitionBinding.RevisionId?.Trim() ?? string.Empty),
RevisionId: resolvedDefinitionBinding.RevisionId?.Trim() ?? string.Empty,
DefinitionVersion: Math.Max(0, resolvedDefinitionBinding.DefinitionVersion)),
wrapAsFallbackTrigger: false,
ct);

Expand Down
3 changes: 3 additions & 0 deletions src/workflow/Aevatar.Workflow.Core/Execution/WorkflowExecutionKernel.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -414,6 +414,7 @@ await TryStartCompensationOrPublishTerminalFailureAsync(
RunId = runId,
Success = false,
Error = evt.Error,
RecoveryFailureKind = evt.RecoveryFailureKind,
},
state,
evt,
Expand Down Expand Up @@ -442,6 +443,7 @@ await TryStartCompensationOrPublishTerminalFailureAsync(
RunId = runId,
Success = false,
Error = evt.Error,
RecoveryFailureKind = evt.RecoveryFailureKind,
},
state,
evt,
Expand Down Expand Up @@ -522,6 +524,7 @@ await TryStartCompensationOrPublishTerminalFailureAsync(
RunId = runId,
Success = false,
Error = WorkflowRuntimeFailureMessages.StepCompletionHandlingFailed(current, evt, ex),
RecoveryFailureKind = evt.RecoveryFailureKind,
},
state,
evt,
Expand Down
6 changes: 6 additions & 0 deletions src/workflow/Aevatar.Workflow.Core/Modules/ConnectorCallModule.ApprovalExecution.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -486,6 +486,12 @@ await WorkflowRuntimeCallbackLeaseSupport.TryCancelAsync(
error,
durationMs,
responseAnnotations);
if (!success &&
string.Equals(reasonCode, "connector_authorization_unavailable", StringComparison.Ordinal))
{
completion.RecoveryFailureKind = WorkflowRecoveryFailureKind.AuthorizationFailure;
}

completion.Annotations["connector.approval.action_id"] = snapshot.Plan.ActionId;
completion.Annotations["connector.approval.status"] = snapshot.ApprovalStatus.ToString();
completion.Annotations["connector.approval.execution_status"] = snapshot.ExecutionStatus.ToString();
Expand Down
36 changes: 35 additions & 1 deletion src/workflow/Aevatar.Workflow.Core/Modules/LLMCallModule.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -182,7 +182,13 @@ private async Task HandleLlmInvocationCompletedAsync(
var publisherActorId = envelope.Route?.PublisherActorId ?? ctx.AgentId;
if (!evt.Success)
{
await PublishFailedCompletionAsync(pending, string.IsNullOrWhiteSpace(evt.Error) ? "LLM call failed." : evt.Error, publisherActorId, ctx, ct);
await PublishFailedCompletionAsync(
pending,
string.IsNullOrWhiteSpace(evt.Error) ? "LLM call failed." : evt.Error,
publisherActorId,
evt.RecoveryFailureKind,
ctx,
ct);
await RemovePendingAsync(sessionId, pending, ctx, ct);
return;
}
Expand Down Expand Up @@ -474,11 +480,37 @@ private static Task PublishFailedCompletionAsync(
string workerId,
IWorkflowExecutionContext ctx,
CancellationToken ct) =>
PublishFailedCompletionAsync(pending, error, workerId, WorkflowRecoveryFailureKind.Unspecified, ctx, ct);

private static Task PublishFailedCompletionAsync(
PendingLlmCallState pending,
string error,
string workerId,
WorkflowRecoveryFailureKind recoveryFailureKind,
IWorkflowExecutionContext ctx,
CancellationToken ct) =>
PublishFailedCompletionAsync(
pending.StepId,
pending.RunId,
error,
workerId,
recoveryFailureKind,
ctx,
ct);

private static Task PublishFailedCompletionAsync(
string stepId,
string runId,
string error,
string workerId,
IWorkflowExecutionContext ctx,
CancellationToken ct) =>
PublishFailedCompletionAsync(
stepId,
runId,
error,
workerId,
WorkflowRecoveryFailureKind.Unspecified,
ctx,
ct);

Expand All @@ -487,6 +519,7 @@ private static Task PublishFailedCompletionAsync(
string runId,
string error,
string workerId,
WorkflowRecoveryFailureKind recoveryFailureKind,
IWorkflowExecutionContext ctx,
CancellationToken ct) =>
ctx.PublishAsync(
Expand All @@ -497,6 +530,7 @@ private static Task PublishFailedCompletionAsync(
Success = false,
Error = error,
WorkerId = string.IsNullOrWhiteSpace(workerId) ? ctx.AgentId : workerId,
RecoveryFailureKind = recoveryFailureKind,
},
TopologyAudience.Self,
ct);
Expand Down
Loading
Loading