package run import ( "crypto/sha256" "encoding/hex" "fmt" "strings" "time" "github.com/tidbcloud/AgentTeam/internal/core/domain" plan "github.com/tidbcloud/AgentTeam/internal/workflow/plan" "github.com/tidbcloud/AgentTeam/internal/workflow/runartifact" ) type Status string const ( StatusQueued Status = "queued" StatusRunning Status = "running" StatusWaiting Status = "waiting" StatusSucceeded Status = "succeeded" StatusFailed Status = "failed" StatusCanceled Status = "canceled" ) func (status Status) Terminal() bool { return status == StatusSucceeded || status == StatusFailed || status == StatusCanceled } type ActivationStatus string const ( ActivationReady ActivationStatus = "ready" ActivationWaiting ActivationStatus = "waiting" ActivationSucceeded ActivationStatus = "succeeded" ActivationFailed ActivationStatus = "failed" ActivationCanceled ActivationStatus = "canceled" ) type ScopeStatus string const ( ScopeActive ScopeStatus = "active" ScopeSucceeded ScopeStatus = "succeeded" ScopeFailed ScopeStatus = "failed" ScopeCanceled ScopeStatus = "canceled" ) type BranchStatus string const ( BranchInactive BranchStatus = "inactive" BranchActive BranchStatus = "active" BranchSucceeded BranchStatus = "succeeded" BranchFailed BranchStatus = "failed" BranchCanceled BranchStatus = "canceled" ) // WorkItemStatus is the durable lifecycle of one frozen fan-out item. Pending // items have not started (either because their dependencies are unresolved or // the concurrency window is full); running covers the complete item lineage, // including durable waits and callbacks. type WorkItemStatus string const ( WorkItemPending WorkItemStatus = "pending" WorkItemRunning WorkItemStatus = "running" WorkItemSucceeded WorkItemStatus = "succeeded" WorkItemFailed WorkItemStatus = "failed" WorkItemBlocked WorkItemStatus = "blocked" WorkItemCanceled WorkItemStatus = "canceled" ) func (status WorkItemStatus) Terminal() bool { switch status { case WorkItemSucceeded, WorkItemFailed, WorkItemBlocked, WorkItemCanceled: return true default: return false } } // Failure is the normalized public-safe failure carried through reducer state, // dispatch, recovery, approval, and callback delivery. Code and Message are the // stable, bounded public problem; Stage/Path/Problem/Hint/Retryable and WorkItem // carry durable diagnostic context. Arbitrary causes, SQL, stacks, credentials, // and tenant identifiers are never authored into this shape. type Failure struct { Code string `json:"code"` Message string `json:"message,omitempty"` Stage string `json:"stage,omitempty"` Path string `json:"path,omitempty"` Problem string `json:"problem,omitempty"` Hint string `json:"hint,omitempty"` Retryable bool `json:"retryable,omitempty"` Diagnostics []FailureDiagnostic `json:"diagnostics,omitempty"` WorkItem *FailureWorkItem `json:"workItem,omitempty"` // Truncation records the deterministic budget cuts already applied to // Diagnostics by an earlier persistence/event boundary. Carrying it forward // on the kernel Failure lets recovery/replay preserve the exact // omittedCount/truncatedFields/serializedBytes instead of re-deriving them // from an already-bounded list (which would recompute the earlier loss away). Truncation *FailureTruncation `json:"truncation,omitempty"` } // FailureDiagnostic is a safe bounded diagnostic (code/path/field/index/message) // carried alongside a failure where a concrete input field or array element is // named. It never carries operational identifiers or arbitrary cause text. type FailureDiagnostic struct { Code string `json:"code,omitempty"` Path string `json:"path,omitempty"` Field string `json:"field,omitempty"` Index *int `json:"index,omitempty"` Message string `json:"message,omitempty"` } // FailureWorkItem identifies the exact dynamic WorkItem (and dependency path) // that produced a fan-out failure, so callers can pinpoint which item's // dependency materialization or settlement failed. type FailureWorkItem struct { Key string `json:"key,omitempty"` Index *int `json:"index,omitempty"` DependencyPath string `json:"dependencyPath,omitempty"` } // ScopeRef is execution lineage, not the artifact's static ScopePath. ScopeID // separates repeated loop iterations and nested instances of the same owner. type ScopeRef struct { ScopeID string `json:"scopeId"` Kind plan.ArtifactScopeKind `json:"kind"` OwnerID string `json:"ownerId"` Branch string `json:"branch,omitempty"` ItemID string `json:"itemId,omitempty"` ItemKey string `json:"itemKey,omitempty"` // ItemIndex is a pointer because source index zero is meaningful while an // omitted value identifies non-fan-out lineage. ItemIndex *int `json:"itemIndex,omitempty"` } type Activation struct { ID string `json:"id"` Sequence uint64 `json:"sequence"` ParentActivationID string `json:"parentActivationId,omitempty"` NodeID string `json:"nodeId"` InputPort string `json:"inputPort"` Input Values `json:"input"` Output Values `json:"output,omitempty"` SelectedPorts []string `json:"selectedPorts"` Status ActivationStatus `json:"status"` Lineage []ScopeRef `json:"lineage"` OwnedScopeIDs []string `json:"ownedScopeIds"` TriggerScopeID string `json:"triggerScopeId,omitempty"` Failure *Failure `json:"failure,omitempty"` // ResourceRejectedCompletion is a compact replay receipt for a completion // whose normalized result could not be admitted into a fan-out item, or // whose causal successor route was redirected through the source's compiled // resource-failed port. The rejected output itself is intentionally not // retained; its deterministic digest preserves same-result idempotency while a // different replay still conflicts. ResourceRejectedCompletion *ResourceRejectedCompletionState `json:"resourceRejectedCompletion,omitempty"` // TaskWatchFrontier contains callback activation IDs representing the // latest Task handlers on every causal control path currently carried by // this activation. Ordinary edges move it, forks copy it, and joins merge // it. CallbackTaskWatch owns the immutable metadata for this activation's // token when it is an eligible fixed-Task callback. TaskWatchFrontier []string `json:"taskWatchFrontier,omitempty"` CallbackTaskWatch *CallbackTaskWatchState `json:"callbackTaskWatch,omitempty"` // FanOutResourceCausal marks an activation whose durable state was reached // from a closing fan-out join. The closing projection follows deterministic // successors up to the first external boundary; this marker keeps the same // exact admission guarantee after that boundary resumes on a later Step or // Complete. FanOutResourceCausal bool `json:"fanOutResourceCausal,omitempty"` } type ResourceRejectedCompletionState struct { Port string `json:"port"` OutputDigest string `json:"outputDigest"` FailureDigest string `json:"failureDigest,omitempty"` CallbackResponderAgentID string `json:"callbackResponderAgentId,omitempty"` } // CallbackTaskWatchState is the durable provenance for one callback token in // the whole-Run Task Watch frontier. ReplacedActivationIDs lets a failed // pre-delivery message restore the prior frontier: the Task transaction never // committed either the successor watch or the predecessor unwatch in that // case. ResponderAgentID is frozen from the committed delivery receipt before // any successor inherits this token. type CallbackTaskWatchState struct { TaskScope domain.MessageScopeValue `json:"taskScope"` ResponderRef string `json:"responderRef"` ResponderAgentID string `json:"responderAgentId,omitempty"` ReplacedActivationIDs []string `json:"replacedActivationIds"` } type BranchState struct { Port string `json:"port"` Status BranchStatus `json:"status"` Value Values `json:"value,omitempty"` Failure *Failure `json:"failure,omitempty"` } type ParallelState struct { ActivationSet string `json:"activationSet"` Mode plan.JoinMode `json:"mode"` JoinNodeID string `json:"joinNodeId"` JoinActivationID string `json:"joinActivationId"` Branches []BranchState `json:"branches"` Outcome string `json:"outcome,omitempty"` WinnerBranch string `json:"winnerBranch,omitempty"` } type LoopState struct { Iteration int `json:"iteration"` Mode plan.LoopMode `json:"mode"` Iterations int `json:"iterations"` State Values `json:"state"` } // WorkItemState is one immutable input plus its mutable execution outcome. // Items remain in their original callback order for the lifetime of the Run. // Direct dependency results are not copied into each consumer. They already // live in immutable terminal WorkItem states and are projected on demand from // DependsOn. This keeps dense DAG state proportional to WorkItems rather than // dependency edges times callback-output size. type WorkItemState struct { ID string `json:"id"` Key string `json:"key"` Index int `json:"index"` Status WorkItemStatus `json:"status"` DependsOn []string `json:"dependsOn"` BlockedBy []string `json:"blockedBy"` Responder string `json:"responder,omitempty"` Responders []string `json:"responders,omitempty"` // SelectOneStrategy is the additive WorkItem selectOne strategy frozen from // the callback value ("" or "ordered" = legacy strict priority, "random" = // injectable unbiased randomization). It is rendered back into the item // responder so the item body's send_message intent carries it to the capacity // scheduler. SelectOneStrategy plan.SelectOneStrategy `json:"selectOneStrategy,omitempty"` WorkScope domain.MessageScopeValue `json:"workScope"` RelatedScopes []domain.MessageScopeValue `json:"relatedScopes"` Instruction string `json:"instruction"` RootActivationID string `json:"rootActivationId,omitempty"` ActivatedResponderAgentIDs []string `json:"activatedResponderAgentIds"` // TaskWatchFrontier is populated only while the WorkItem itself is the // causal carrier: before a direct item result enters a body, after a body // arrives, or after a body fails. Running item bodies carry the same tokens // on their activation frontier instead. TaskWatchFrontier []string `json:"taskWatchFrontier,omitempty"` Value Values `json:"value"` Failure *Failure `json:"failure,omitempty"` } // FanOutState owns the dynamic instances for one fan_out activation. The // compiler still owns a single static body and paired join; only these frozen // item instances are dynamic. type FanOutState struct { OwnerActivationID string `json:"ownerActivationId"` JoinNodeID string `json:"joinNodeId"` JoinActivationID string `json:"joinActivationId"` MaxConcurrency int `json:"maxConcurrency"` FailureMode plan.FanOutFailureMode `json:"failureMode"` Items []WorkItemState `json:"items"` Outcome string `json:"outcome,omitempty"` // ResourceFailure is set only when the authored completed settlement cannot // fit even after compacting the currently closing WorkItem. It preserves all // WorkItem outcomes and makes the paired join take its recoverable failed // port instead of rewriting an unrelated successful item. ResourceFailure bool `json:"resourceFailure,omitempty"` } type Scope struct { ID string `json:"id"` Sequence uint64 `json:"sequence"` Kind plan.ArtifactScopeKind `json:"kind"` OwnerID string `json:"ownerId"` Status ScopeStatus `json:"status"` ParentLineage []ScopeRef `json:"parentLineage"` Parallel *ParallelState `json:"parallel,omitempty"` Loop *LoopState `json:"loop,omitempty"` FanOut *FanOutState `json:"fanOut,omitempty"` } // State is the entire pure graph snapshot. Operational leases, attempts, // receipts and event sequence allocation remain store concerns. type State struct { Status Status `json:"status"` Input Values `json:"input"` Variables Values `json:"variables"` Output Values `json:"output,omitempty"` Failure *Failure `json:"failure,omitempty"` Activations []Activation `json:"activations"` Scopes []Scope `json:"scopes"` NextActivation uint64 `json:"nextActivation"` NextScope uint64 `json:"nextScope"` } type IntentKind string const ( IntentSendMessage IntentKind = "send_message" IntentWait IntentKind = "wait" IntentApproval IntentKind = "request_approval" ) // WorkScopeMode is the compiler-owned collaboration-scope mode carried into // the durable effect protocol. type WorkScopeMode = plan.ArtifactWorkScopeMode const ( WorkScopeModeAgent = plan.ArtifactWorkScopeAgent WorkScopeModeFixed = plan.ArtifactWorkScopeFixed ) // Intent is a durable side-effect/timer request. Stable ActivationID is the // semantic idempotency basis; delivery attempts must not create new intents. type Intent struct { Kind IntentKind `json:"kind"` ActivationID string `json:"activationId"` NodeID string `json:"nodeId"` Target string `json:"target,omitempty"` Scope *domain.MessageScopeValue `json:"scope,omitempty"` Responder string `json:"responder,omitempty"` // ResponderCandidates is the frozen ordered candidate pool of // responder.selectOne. It is non-empty exactly when the effect declared a // selectOne pool instead of a fixed responder; order is the runtime // priority when several candidates are simultaneously available (unless // SelectOneStrategyRandom is set, which randomizes over currently available // deduplicated candidates). It is the durable, lossless handoff to the // capacity scheduler. ResponderCandidates []string `json:"responderCandidates,omitempty"` // SelectOneStrategy is the additive responder.selectOne strategy carried from // the Plan artifact ("" or "ordered" = legacy strict priority, "random" = // injectable unbiased randomization over currently available deduplicated // candidates). SelectOneStrategy plan.SelectOneStrategy `json:"selectOneStrategy,omitempty"` Text string `json:"text,omitempty"` // TextRef is the compact immutable artifact binding used when rendered // message content would exceed one effect request. It is resolved only by // the trusted dispatcher; client callbacks cannot supply it. TextRef *runartifact.Ref `json:"textRef,omitempty"` Callback bool `json:"callback,omitempty"` StartThread bool `json:"startThread,omitempty"` WorkScopeMode WorkScopeMode `json:"workScopeMode,omitempty"` WorkScope *domain.MessageScopeValue `json:"workScope,omitempty"` RelatedScopes []domain.MessageScopeValue `json:"relatedScopes,omitempty"` TaskWatchHandoff *TaskWatchHandoff `json:"taskWatchHandoff,omitempty"` CallbackOutput plan.ArtifactContract `json:"callbackOutput,omitempty"` // Nil means this callback does not support WorkItems. A non-nil slice is // the normalized, frozen responder allowlist for callback validation. WorkItemsAllowedResponders []string `json:"workItemsAllowedResponders,omitempty"` // WorkItemsDependencyValueContract is non-nil exactly when WorkItems are // enabled. It is frozen from the verified downstream fan_out Join item // input and describes the object stored in every dependency envelope's // value field. A pointer preserves an intentionally empty [] contract while // leaving v1-v3 Intent bytes unchanged. WorkItemsDependencyValueContract *plan.ArtifactContract `json:"workItemsDependencyValueContract,omitempty"` Assignee string `json:"assignee,omitempty"` Summary string `json:"summary,omitempty"` // FeedbackScope is the optional canonical collaboration scope frozen when // request_approval activates. It is never re-evaluated from mutable Plan // source while the approval is awaiting or being decided. FeedbackScope *domain.MessageScopeValue `json:"feedbackScope,omitempty"` WakeAt *time.Time `json:"wakeAt,omitempty"` Deadline *time.Time `json:"deadline,omitempty"` Retry plan.ArtifactRetryPolicy `json:"retry"` } // TaskWatchHandoff is the immutable Task Watch transition attached to one // structured callback with matching fixed Task delivery/work scopes. The // current responder is resolved normalized by the ordinary message // transaction. For a fan-out WorkItem, predecessor IDs were already frozen by // their callback receipts, so an effect replay never reinterprets mutable // Agent names or WorkItem dependencies. type TaskWatchHandoff struct { TaskScope domain.MessageScopeValue `json:"taskScope"` // WorkItemID is the concrete WorkItem ID inside fan_out and the stable // activation ID for an ordinary static callback. WorkItemID string `json:"workItemId"` CausalPredecessorAgentIDs []string `json:"causalPredecessorAgentIds"` } // Event is a durable public-safe Run event. Failure events (run_failed, // branch_failed, fan_out_item_failed) carry the public safe failure contract // (code/stage/path/problem/hint/retryable/workItem/diagnostics/truncation) so // the persisted event payload and its Run log Details projection preserve every // field the Agent/CLI needs to act on the failure. The raw authoring message // stays on the kernel Failure and never crosses into the event payload. type Event struct { Kind string `json:"kind"` ActivationID string `json:"activationId,omitempty"` NodeID string `json:"nodeId,omitempty"` AgentID string `json:"agentId,omitempty"` Port string `json:"port,omitempty"` ScopeID string `json:"scopeId,omitempty"` Code string `json:"code,omitempty"` Stage string `json:"stage,omitempty"` Path string `json:"path,omitempty"` Problem string `json:"problem,omitempty"` Hint string `json:"hint,omitempty"` Retryable bool `json:"retryable,omitempty"` WorkItem *FailureWorkItem `json:"workItem,omitempty"` Diagnostics []FailureDiagnostic `json:"diagnostics,omitempty"` // Truncation reports the deterministic budget cuts the kernel FailureEvent // applied to Diagnostics (count cap, per-field rune cap, serialized-size // cap) so the persisted event payload and its Run log Details projection // carry the exact same omittedCount/truncatedFields/serializedBytes as the // error_json. It is populated only when a cut fired. Truncation *FailureTruncation `json:"truncation,omitempty"` } // FailureEvent builds a public safe failure Event from a kernel Failure. The // raw authoring Message is never projected; only the bounded public-safe fields // are carried so the persisted event payload and its Run log Details projection // preserve code/stage/path/problem/hint/retryable/workItem/diagnostics and the // deterministic diagnostic budget cuts (truncation metadata). func FailureEvent(kind string, failure *Failure, scopeID string, activationID string, nodeID string) Event { event := Event{ Kind: kind, ScopeID: strings.TrimSpace(scopeID), ActivationID: strings.TrimSpace(activationID), NodeID: strings.TrimSpace(nodeID), } if failure == nil { return event } // Identity fields (code/stage/path) are validated byte-for-byte and fail // closed to empty when malformed; they are never TrimSpace'd or sanitized // into a different valid identity. Problem/hint are free text and may trim. event.Code, _ = PublicFailureCode(failure.Code) event.Stage, _ = PublicFailureStage(failure.Stage) event.Path, _ = PublicFailurePath(failure.Path) event.Problem = strings.TrimSpace(failure.Problem) event.Hint = strings.TrimSpace(failure.Hint) event.Retryable = failure.Retryable // Always route through the shared carried budget: even an empty diagnostics // list carries a prior truncation report forward so a replayed/all-dropped // failure preserves omittedCount/truncatedFields/serializedBytes. event.Diagnostics, event.Truncation = BoundFailureDiagnosticsCarried(failure.Diagnostics, failure.Truncation) if failure.WorkItem != nil { workItem := *failure.WorkItem event.WorkItem = &workItem } return event } type Transition struct { State *State `json:"state"` Intents []Intent `json:"intents"` Events []Event `json:"events"` } type Completion struct { ActivationID string `json:"activationId"` Port string `json:"port"` Output Values `json:"output"` Failure *Failure `json:"failure,omitempty"` CallbackResponderAgentID string `json:"callbackResponderAgentId,omitempty"` } func (state *State) Clone() *State { if state == nil { return nil } copyState := *state copyState.Input = cloneValues(state.Input) copyState.Variables = cloneValues(state.Variables) copyState.Output = cloneValues(state.Output) copyState.Failure = cloneFailure(state.Failure) copyState.Activations = make([]Activation, len(state.Activations)) for index := range state.Activations { copyState.Activations[index] = cloneActivation(state.Activations[index]) } copyState.Scopes = make([]Scope, len(state.Scopes)) for index := range state.Scopes { copyState.Scopes[index] = cloneScope(state.Scopes[index]) } return ©State } func cloneActivation(activation Activation) Activation { activation.Input = cloneValues(activation.Input) activation.Output = cloneValues(activation.Output) activation.SelectedPorts = append([]string(nil), activation.SelectedPorts...) activation.Lineage = cloneScopeRefs(activation.Lineage) activation.OwnedScopeIDs = append([]string(nil), activation.OwnedScopeIDs...) activation.TaskWatchFrontier = append([]string(nil), activation.TaskWatchFrontier...) if activation.CallbackTaskWatch != nil { callbackTaskWatch := *activation.CallbackTaskWatch callbackTaskWatch.ReplacedActivationIDs = append([]string{}, activation.CallbackTaskWatch.ReplacedActivationIDs...) activation.CallbackTaskWatch = &callbackTaskWatch } activation.Failure = cloneFailure(activation.Failure) if activation.ResourceRejectedCompletion != nil { rejected := *activation.ResourceRejectedCompletion activation.ResourceRejectedCompletion = &rejected } return activation } func cloneScope(scope Scope) Scope { scope.ParentLineage = cloneScopeRefs(scope.ParentLineage) if scope.Parallel != nil { parallel := *scope.Parallel parallel.Branches = make([]BranchState, len(scope.Parallel.Branches)) for index, branch := range scope.Parallel.Branches { branch.Value = cloneValues(branch.Value) branch.Failure = cloneFailure(branch.Failure) parallel.Branches[index] = branch } scope.Parallel = ¶llel } if scope.Loop != nil { loop := *scope.Loop loop.State = cloneValues(loop.State) scope.Loop = &loop } if scope.FanOut != nil { fanOut := *scope.FanOut fanOut.Items = make([]WorkItemState, len(scope.FanOut.Items)) for index, item := range scope.FanOut.Items { // These are closed array fields in the durable state schema. Keep // empty arrays non-nil so cloning and JSON recovery do not silently // turn [] into null. item.DependsOn = append([]string{}, item.DependsOn...) item.BlockedBy = append([]string{}, item.BlockedBy...) item.RelatedScopes = append([]domain.MessageScopeValue{}, item.RelatedScopes...) item.ActivatedResponderAgentIDs = append([]string{}, item.ActivatedResponderAgentIDs...) item.TaskWatchFrontier = append([]string(nil), item.TaskWatchFrontier...) item.Value = cloneValues(item.Value) item.Failure = cloneFailure(item.Failure) fanOut.Items[index] = item } scope.FanOut = &fanOut } return scope } func cloneScopeRefs(references []ScopeRef) []ScopeRef { result := make([]ScopeRef, len(references)) for index, reference := range references { if reference.ItemIndex != nil { itemIndex := *reference.ItemIndex reference.ItemIndex = &itemIndex } result[index] = reference } return result } func cloneFailure(failure *Failure) *Failure { if failure == nil { return nil } result := *failure if failure.Truncation != nil { truncation := *failure.Truncation truncation.TruncatedFields = append([]string(nil), failure.Truncation.TruncatedFields...) result.Truncation = &truncation } return &result } func (state *State) activation(id string) (*Activation, error) { for index := range state.Activations { if state.Activations[index].ID == id { return &state.Activations[index], nil } } return nil, fmt.Errorf("unknown activation %q", id) } func (state *State) scope(id string) (*Scope, error) { for index := range state.Scopes { if state.Scopes[index].ID == id { return &state.Scopes[index], nil } } return nil, fmt.Errorf("unknown scope %q", id) } func (state *State) appendActivation(nodeID, port string, input Values, lineage []ScopeRef, status ActivationStatus) (*Activation, error) { if len(state.Activations) >= plan.MaxRunActivations || state.NextActivation >= uint64(plan.MaxRunActivations) { return nil, fmt.Errorf("workflow Run cannot exceed %d activations", plan.MaxRunActivations) } state.NextActivation++ activation := Activation{ ID: fmt.Sprintf("a%020d", state.NextActivation), Sequence: state.NextActivation, NodeID: nodeID, InputPort: port, Input: cloneValues(input), SelectedPorts: []string{}, Status: status, Lineage: cloneScopeRefs(lineage), OwnedScopeIDs: []string{}, } state.Activations = append(state.Activations, activation) return &state.Activations[len(state.Activations)-1], nil } func (state *State) appendScope(kind plan.ArtifactScopeKind, ownerID string, lineage []ScopeRef) *Scope { state.NextScope++ scope := Scope{ ID: fmt.Sprintf("s%020d", state.NextScope), Sequence: state.NextScope, Kind: kind, OwnerID: ownerID, Status: ScopeActive, ParentLineage: cloneScopeRefs(lineage), } state.Scopes = append(state.Scopes, scope) return &state.Scopes[len(state.Scopes)-1] } // WorkItemSemanticID deterministically identifies one item inside the fan-out // owner activation. The Run aggregate is the namespace boundary; callers that // need a globally unique storage ID must additionally domain-separate by Run // ID in the projection layer. func WorkItemSemanticID(ownerActivationID, key string) string { digest := sha256.Sum256([]byte("loop.workflow.work-item.v1\x00" + ownerActivationID + "\x00" + key)) return "wfi_" + hex.EncodeToString(digest[:16]) }