weichen
6a3a0dc201
- FIND-SEC-01: GetUserContext requires session ownership (handler 403 + SQL/memory owner filter); in-memory HTTP + isolated-PG regressions - FIND-ARCH-01: archtest fails closed on untracked table writes (+noise exemption) and registers 50 existing writer tables in table_owners - FIND-SEC-02: Authorizer without pool denies by default; explicit NewInMemoryAuthorizer seam opt-in via ALLOW_IN_MEMORY at composition - FIND-SEC-03: swboot commands run argv-exec (no sh -c) with exact basename allowlist match; prefix-spoof and no-shell-eval tests - FIND-FLOW-01: confirm rejects validated_only plans; ExecuteClaimed hard-refuses validated_only steps (new error codes + HTTP mapping) - FIND-QUAL-01: intentflow.ValidateFlagConsistency fail-fast at Assemble
745 lines
20 KiB
Go
745 lines
20 KiB
Go
package intentflow
|
||
|
||
import (
|
||
"context"
|
||
"crypto/rand"
|
||
"encoding/hex"
|
||
"errors"
|
||
"fmt"
|
||
"slices"
|
||
"time"
|
||
|
||
"chat-agent-system/internal/app/uow"
|
||
)
|
||
|
||
const (
|
||
defaultLeaseTTL = 15 * time.Second
|
||
defaultPlanTTL = 30 * time.Minute
|
||
)
|
||
|
||
// Plans is the owner implementation of Service plus worker/recovery methods.
|
||
type Plans struct {
|
||
uow uow.UnitOfWork
|
||
store Store
|
||
owner ActionOwner
|
||
authz StepAuthorizer
|
||
events EventSink
|
||
bind ResourceVersionBinder
|
||
now func() time.Time
|
||
newID func(prefix string) string
|
||
lease time.Duration
|
||
ttl time.Duration
|
||
}
|
||
|
||
// ResourceVersionBindInput carries enough context to resolve optimistic versions
|
||
// for mutate/docgov steps at plan-create time (U10/U11).
|
||
type ResourceVersionBindInput struct {
|
||
Principal Principal
|
||
ActionKey string
|
||
SafeSummary string
|
||
Slots []byte
|
||
}
|
||
|
||
// ResourceVersionBinder returns the expected resource version for a draft step.
|
||
// Empty string means the action key does not require a version binding.
|
||
type ResourceVersionBinder func(ctx context.Context, in ResourceVersionBindInput) (string, error)
|
||
|
||
type Options struct {
|
||
Owner ActionOwner
|
||
Authorizer StepAuthorizer
|
||
Events EventSink
|
||
ResourceVersionBinder ResourceVersionBinder
|
||
Now func() time.Time
|
||
NewID func(prefix string) string
|
||
Lease time.Duration
|
||
PlanTTL time.Duration
|
||
}
|
||
|
||
func NewPlans(unit uow.UnitOfWork, store Store, opt Options) *Plans {
|
||
now := opt.Now
|
||
if now == nil {
|
||
now = func() time.Time { return time.Now().UTC() }
|
||
}
|
||
newID := opt.NewID
|
||
if newID == nil {
|
||
newID = randomID
|
||
}
|
||
lease := opt.Lease
|
||
if lease <= 0 {
|
||
lease = defaultLeaseTTL
|
||
}
|
||
ttl := opt.PlanTTL
|
||
if ttl <= 0 {
|
||
ttl = defaultPlanTTL
|
||
}
|
||
authz := opt.Authorizer
|
||
if authz == nil {
|
||
if a, ok := opt.Owner.(StepAuthorizer); ok {
|
||
authz = a
|
||
}
|
||
}
|
||
return &Plans{
|
||
uow: unit, store: store, owner: opt.Owner, authz: authz, events: opt.Events,
|
||
bind: opt.ResourceVersionBinder, now: now, newID: newID, lease: lease, ttl: ttl,
|
||
}
|
||
}
|
||
|
||
func (p *Plans) stepAuthorizer() StepAuthorizer {
|
||
if p == nil {
|
||
return nil
|
||
}
|
||
if p.authz != nil {
|
||
return p.authz
|
||
}
|
||
if a, ok := p.owner.(StepAuthorizer); ok {
|
||
return a
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func randomID(prefix string) string {
|
||
var b [16]byte
|
||
if _, err := rand.Read(b[:]); err != nil {
|
||
return prefix + fmt.Sprintf("%d", time.Now().UnixNano())
|
||
}
|
||
return prefix + hex.EncodeToString(b[:])
|
||
}
|
||
|
||
func requirePrincipal(p Principal) error {
|
||
if p.UserID <= 0 || p.ProjectID <= 0 || p.SessionID == "" || p.Environment == "" {
|
||
return ErrUnauthenticated
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func visibleTo(plan *Plan, p Principal) bool {
|
||
return plan != nil && sameScope(plan.Principal, p)
|
||
}
|
||
|
||
func (p *Plans) GetPlan(ctx context.Context, principal Principal, planID string) (*PlanView, error) {
|
||
if err := requirePrincipal(principal); err != nil {
|
||
return nil, err
|
||
}
|
||
var view *PlanView
|
||
err := p.withTx(ctx, func(ctx context.Context, scope uow.TxScope) error {
|
||
plan, err := p.store.GetPlan(ctx, scope, planID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !visibleTo(plan, principal) {
|
||
return ErrPlanNotFoundOrHidden
|
||
}
|
||
view = toView(plan)
|
||
return nil
|
||
})
|
||
return view, err
|
||
}
|
||
|
||
func (p *Plans) GetActivePlan(ctx context.Context, principal Principal) (*PlanView, error) {
|
||
if err := requirePrincipal(principal); err != nil {
|
||
return nil, err
|
||
}
|
||
var view *PlanView
|
||
err := p.withTx(ctx, func(ctx context.Context, scope uow.TxScope) error {
|
||
plan, err := p.store.GetActive(ctx, scope, principal)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if plan == nil {
|
||
return nil
|
||
}
|
||
view = toView(plan)
|
||
return nil
|
||
})
|
||
return view, err
|
||
}
|
||
|
||
func (p *Plans) CreatePlan(ctx context.Context, in CreateCommand) (*PlanView, error) {
|
||
if err := requirePrincipal(in.Principal); err != nil {
|
||
return nil, err
|
||
}
|
||
if in.ClientTurnID == "" || in.RequestDigest == "" {
|
||
return nil, ErrPlanDecisionConflict
|
||
}
|
||
if len(in.Steps) == 0 || len(in.Steps) > 4 {
|
||
return nil, ErrPlanDependencyInvalid
|
||
}
|
||
now := p.now()
|
||
status := in.Status
|
||
if status == "" {
|
||
status = PlanAwaitingConfirmation
|
||
}
|
||
mode := in.ExecutionMode
|
||
if mode == "" {
|
||
mode = ModeValidatedOnly
|
||
}
|
||
expires := in.ExpiresAt
|
||
if expires.IsZero() {
|
||
expires = now.Add(p.ttl)
|
||
}
|
||
seg := in.SegmenterVersion
|
||
if seg == "" {
|
||
seg = SegmenterVersionUnknown
|
||
}
|
||
pol := in.PolicyVersion
|
||
if pol == "" {
|
||
pol = PolicyVersion
|
||
}
|
||
|
||
var view *PlanView
|
||
err := p.withTx(ctx, func(ctx context.Context, scope uow.TxScope) error {
|
||
existing, err := p.store.GetByTurn(ctx, scope, in.Principal.Environment, in.Principal.SessionID, in.ClientTurnID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if existing != nil {
|
||
if existing.RequestDigest != in.RequestDigest {
|
||
return ErrPlanDecisionConflict
|
||
}
|
||
view = toView(existing)
|
||
return nil
|
||
}
|
||
if in.WritePlan {
|
||
active, err := p.store.GetActiveWrite(ctx, scope, in.Principal)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if active != nil {
|
||
view = toView(active)
|
||
return ErrPlanDecisionConflict
|
||
}
|
||
}
|
||
plan := &Plan{
|
||
PlanID: p.newID("pln_"),
|
||
Principal: in.Principal,
|
||
ClientTurnID: in.ClientTurnID,
|
||
RequestDigest: in.RequestDigest,
|
||
SegmenterVersion: seg,
|
||
PolicyVersion: pol,
|
||
Status: status,
|
||
Revision: 1,
|
||
ExpiresAt: expires,
|
||
WritePlan: in.WritePlan,
|
||
ExecutionMode: mode,
|
||
SafeSummary: in.SafeSummary,
|
||
CreatedAt: now,
|
||
UpdatedAt: now,
|
||
}
|
||
for i, d := range in.Steps {
|
||
mode := d.ExecutionMode
|
||
if mode == "" {
|
||
mode = plan.ExecutionMode
|
||
}
|
||
slots := d.Slots
|
||
if len(slots) == 0 {
|
||
slots = []byte("{}")
|
||
}
|
||
plan.Steps = append(plan.Steps, Step{
|
||
PlanID: plan.PlanID,
|
||
StepID: p.newID("stp_"),
|
||
Ordinal: i + 1,
|
||
ActionKey: d.ActionKey,
|
||
Slots: slots,
|
||
DependsOn: slices.Clone(d.DependsOn),
|
||
Status: StepPending,
|
||
ExpectedResourceVersion: d.ExpectedResourceVersion,
|
||
ActionDigest: d.ActionDigest,
|
||
SafeSummary: d.SafeSummary,
|
||
Version: 1,
|
||
ExecutionMode: mode,
|
||
CreatedAt: now,
|
||
UpdatedAt: now,
|
||
})
|
||
}
|
||
if err := p.store.InsertPlan(ctx, scope, plan); err != nil {
|
||
if errors.Is(err, errTurnExists) {
|
||
again, loadErr := p.store.GetByTurn(ctx, scope, in.Principal.Environment, in.Principal.SessionID, in.ClientTurnID)
|
||
if loadErr != nil {
|
||
return loadErr
|
||
}
|
||
if again != nil && again.RequestDigest == in.RequestDigest {
|
||
view = toView(again)
|
||
return nil
|
||
}
|
||
return ErrPlanDecisionConflict
|
||
}
|
||
if errors.Is(err, uow.ErrBusy) {
|
||
active, loadErr := p.store.GetActiveWrite(ctx, scope, in.Principal)
|
||
if loadErr != nil {
|
||
return loadErr
|
||
}
|
||
if active != nil {
|
||
view = toView(active)
|
||
}
|
||
return ErrPlanDecisionConflict
|
||
}
|
||
return err
|
||
}
|
||
view = toView(plan)
|
||
return p.emitFact(ctx, plan, "", now)
|
||
})
|
||
return view, err
|
||
}
|
||
|
||
func (p *Plans) DecidePlan(ctx context.Context, in DecisionCommand) (*PlanView, error) {
|
||
if err := requirePrincipal(in.Principal); err != nil {
|
||
return nil, err
|
||
}
|
||
if in.PlanID == "" || in.DecisionNonce == "" || in.DecisionDigest == "" {
|
||
return nil, ErrPlanDecisionConflict
|
||
}
|
||
now := p.now()
|
||
var view *PlanView
|
||
err := p.withTx(ctx, func(ctx context.Context, scope uow.TxScope) error {
|
||
plan, err := p.store.GetPlanForUpdate(ctx, scope, in.PlanID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !visibleTo(plan, in.Principal) {
|
||
return ErrPlanNotFoundOrHidden
|
||
}
|
||
existing, err := p.store.GetDecisionByNonce(ctx, scope, in.PlanID, in.DecisionNonce)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if existing != nil {
|
||
if existing.DecisionDigest != in.DecisionDigest {
|
||
return ErrPlanDecisionConflict
|
||
}
|
||
fresh, err := p.store.GetPlan(ctx, scope, in.PlanID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
view = toView(fresh)
|
||
return nil
|
||
}
|
||
if isWaitingStatus(plan.Status) && !now.Before(plan.ExpiresAt) {
|
||
if _, err := p.store.CASPlan(ctx, scope, plan.PlanID, plan.Revision, waitingStatuses(), PlanExpired, now); err != nil {
|
||
return err
|
||
}
|
||
cancelUnstarted(plan, now)
|
||
if err := p.store.ReplaceSteps(ctx, scope, plan.PlanID, plan.Steps); err != nil {
|
||
return err
|
||
}
|
||
return ErrPlanExpired
|
||
}
|
||
if in.ExpectedRevision != plan.Revision {
|
||
return ErrPlanStaleRevision
|
||
}
|
||
if err := p.authorizeDecision(ctx, plan, in); err != nil {
|
||
return err
|
||
}
|
||
dec := &Decision{
|
||
DecisionID: p.newID("dec_"),
|
||
PlanID: plan.PlanID,
|
||
Nonce: in.DecisionNonce,
|
||
PlanRevision: plan.Revision,
|
||
ActorUserID: in.Principal.UserID,
|
||
Kind: in.Kind,
|
||
StepIDs: slices.Clone(in.StepIDs),
|
||
ExcludeUnselected: in.ExcludeUnselected,
|
||
DecisionDigest: in.DecisionDigest,
|
||
CreatedAt: now,
|
||
}
|
||
if err := p.store.InsertDecision(ctx, scope, dec); err != nil {
|
||
if errors.Is(err, uow.ErrDigestConflict) || errors.Is(err, errTurnExists) {
|
||
again, loadErr := p.store.GetDecisionByNonce(ctx, scope, in.PlanID, in.DecisionNonce)
|
||
if loadErr != nil {
|
||
return loadErr
|
||
}
|
||
if again != nil && again.DecisionDigest == in.DecisionDigest {
|
||
fresh, err := p.store.GetPlan(ctx, scope, in.PlanID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
view = toView(fresh)
|
||
return nil
|
||
}
|
||
return ErrPlanDecisionConflict
|
||
}
|
||
return err
|
||
}
|
||
from := []PlanStatus{plan.Status}
|
||
var to PlanStatus
|
||
switch in.Kind {
|
||
case DecisionConfirm, DecisionConfirmRetry, DecisionContinue:
|
||
if in.Kind == DecisionConfirm {
|
||
// FIND-FLOW-01 回归：validated_only（shadow/dry-run）计划不允许
|
||
// 经 Confirm 进入 executing；shadow 安全不依赖「没人点 Confirm」。
|
||
if plan.ExecutionMode == ModeValidatedOnly {
|
||
return ErrPlanModeNotConfirmable
|
||
}
|
||
if plan.Status != PlanAwaitingConfirmation && plan.Status != PlanAwaitingClarification && plan.Status != PlanProposed {
|
||
return ErrPlanStaleRevision
|
||
}
|
||
} else if plan.Status != PlanPartiallyCompleted && plan.Status != PlanExecuting {
|
||
return ErrPlanStaleRevision
|
||
}
|
||
if err := applyConfirm(plan, in.StepIDs, in.ExcludeUnselected, now); err != nil {
|
||
return err
|
||
}
|
||
to = PlanExecuting
|
||
case DecisionCancel:
|
||
if !isWaitingStatus(plan.Status) && plan.Status != PlanExecuting && plan.Status != PlanPartiallyCompleted {
|
||
return ErrPlanStaleRevision
|
||
}
|
||
cancelUnstarted(plan, now)
|
||
to = PlanCancelled
|
||
case DecisionExclude:
|
||
if !isWaitingStatus(plan.Status) && plan.Status != PlanPartiallyCompleted {
|
||
return ErrPlanStaleRevision
|
||
}
|
||
if err := applyExclude(plan, in.StepIDs, now); err != nil {
|
||
return err
|
||
}
|
||
to = plan.Status
|
||
if allStepsTerminal(plan) {
|
||
to = PlanCancelled
|
||
}
|
||
case DecisionEdit:
|
||
if !isWaitingStatus(plan.Status) {
|
||
return ErrPlanStaleRevision
|
||
}
|
||
to = plan.Status
|
||
default:
|
||
return ErrPlanDependencyInvalid
|
||
}
|
||
newRev, err := p.store.CASPlan(ctx, scope, plan.PlanID, plan.Revision, from, to, now)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if newRev == 0 {
|
||
return ErrPlanStaleRevision
|
||
}
|
||
if err := p.store.ReplaceSteps(ctx, scope, plan.PlanID, plan.Steps); err != nil {
|
||
return err
|
||
}
|
||
plan.Status = to
|
||
plan.Revision = newRev
|
||
view = toView(plan)
|
||
return p.emitFact(ctx, plan, "", now)
|
||
})
|
||
return view, err
|
||
}
|
||
|
||
func (p *Plans) authorizeDecision(ctx context.Context, plan *Plan, in DecisionCommand) error {
|
||
switch in.Kind {
|
||
case DecisionConfirm, DecisionConfirmRetry, DecisionContinue:
|
||
default:
|
||
return nil
|
||
}
|
||
auth := p.stepAuthorizer()
|
||
if auth == nil {
|
||
return nil
|
||
}
|
||
ids, err := selectedStepIDs(plan, in.StepIDs)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
byID := map[string]Step{}
|
||
for _, s := range plan.Steps {
|
||
byID[s.StepID] = s
|
||
}
|
||
for _, id := range ids {
|
||
s := byID[id]
|
||
if err := auth.Authorize(ctx, StepExec{
|
||
Principal: in.Principal,
|
||
PlanID: plan.PlanID,
|
||
StepID: s.StepID,
|
||
ActionKey: s.ActionKey,
|
||
ActionDigest: s.ActionDigest,
|
||
ExpectedResourceVersion: s.ExpectedResourceVersion,
|
||
Slots: s.Slots,
|
||
SafeSummary: s.SafeSummary,
|
||
}); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func selectedStepIDs(plan *Plan, selected []string) ([]string, error) {
|
||
byID := map[string]*Step{}
|
||
for i := range plan.Steps {
|
||
byID[plan.Steps[i].StepID] = &plan.Steps[i]
|
||
}
|
||
if len(selected) == 0 {
|
||
ids := make([]string, 0, len(plan.Steps))
|
||
for i := range plan.Steps {
|
||
ids = append(ids, plan.Steps[i].StepID)
|
||
}
|
||
return ids, nil
|
||
}
|
||
chosen := map[string]bool{}
|
||
for _, id := range selected {
|
||
if _, ok := byID[id]; !ok {
|
||
return nil, ErrPlanDependencyInvalid
|
||
}
|
||
chosen[id] = true
|
||
}
|
||
for id := range chosen {
|
||
for _, dep := range byID[id].DependsOn {
|
||
depStep, ok := byID[dep]
|
||
if !ok || (!chosen[dep] && depStep.Status != StepSucceeded && depStep.Status != StepSkipped) {
|
||
return nil, ErrPlanDependencyInvalid
|
||
}
|
||
}
|
||
}
|
||
ids := make([]string, 0, len(chosen))
|
||
for id := range chosen {
|
||
ids = append(ids, id)
|
||
}
|
||
return ids, nil
|
||
}
|
||
|
||
func applyExclude(plan *Plan, stepIDs []string, now time.Time) error {
|
||
if len(stepIDs) == 0 {
|
||
return ErrPlanDependencyInvalid
|
||
}
|
||
byID := map[string]*Step{}
|
||
for i := range plan.Steps {
|
||
byID[plan.Steps[i].StepID] = &plan.Steps[i]
|
||
}
|
||
for _, id := range stepIDs {
|
||
s, ok := byID[id]
|
||
if !ok {
|
||
return ErrPlanDependencyInvalid
|
||
}
|
||
if s.Status == StepPending || s.Status == StepEligible {
|
||
s.Status = StepCancelled
|
||
s.UpdatedAt = now
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func allStepsTerminal(plan *Plan) bool {
|
||
for _, s := range plan.Steps {
|
||
switch s.Status {
|
||
case StepSucceeded, StepFailed, StepDenied, StepSkipped, StepCancelled:
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
return true
|
||
}
|
||
|
||
func applyConfirm(plan *Plan, selected []string, excludeUnselected bool, now time.Time) error {
|
||
byID := map[string]*Step{}
|
||
for i := range plan.Steps {
|
||
byID[plan.Steps[i].StepID] = &plan.Steps[i]
|
||
}
|
||
chosen := map[string]bool{}
|
||
if len(selected) == 0 {
|
||
for i := range plan.Steps {
|
||
chosen[plan.Steps[i].StepID] = true
|
||
}
|
||
} else {
|
||
for _, id := range selected {
|
||
if _, ok := byID[id]; !ok {
|
||
return ErrPlanDependencyInvalid
|
||
}
|
||
chosen[id] = true
|
||
}
|
||
for id := range chosen {
|
||
for _, dep := range byID[id].DependsOn {
|
||
depStep, ok := byID[dep]
|
||
if !ok || (!chosen[dep] && depStep.Status != StepSucceeded && depStep.Status != StepSkipped) {
|
||
return ErrPlanDependencyInvalid
|
||
}
|
||
}
|
||
}
|
||
}
|
||
for i := range plan.Steps {
|
||
s := &plan.Steps[i]
|
||
if chosen[s.StepID] {
|
||
if s.Status == StepPending || s.Status == StepEligible {
|
||
s.Status = StepEligible
|
||
s.UpdatedAt = now
|
||
}
|
||
continue
|
||
}
|
||
if excludeUnselected && (s.Status == StepPending || s.Status == StepEligible) {
|
||
s.Status = StepCancelled
|
||
s.UpdatedAt = now
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func cancelUnstarted(plan *Plan, now time.Time) {
|
||
for i := range plan.Steps {
|
||
s := &plan.Steps[i]
|
||
if s.Status == StepPending || s.Status == StepEligible {
|
||
s.Status = StepCancelled
|
||
s.UpdatedAt = now
|
||
}
|
||
}
|
||
}
|
||
|
||
func (p *Plans) ClaimStep(ctx context.Context, workerID, planID, stepID string) (*ClaimedStep, error) {
|
||
if workerID == "" || planID == "" || stepID == "" {
|
||
return nil, ErrPlanDependencyInvalid
|
||
}
|
||
now := p.now()
|
||
until := now.Add(p.lease)
|
||
var out *ClaimedStep
|
||
err := p.withTx(ctx, func(ctx context.Context, scope uow.TxScope) error {
|
||
plan, err := p.store.GetPlanForUpdate(ctx, scope, planID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if plan == nil {
|
||
return ErrPlanNotFoundOrHidden
|
||
}
|
||
var digest string
|
||
for _, s := range plan.Steps {
|
||
if s.StepID == stepID {
|
||
digest = s.ActionDigest
|
||
break
|
||
}
|
||
}
|
||
key := fmt.Sprintf("%s:%d:%s:%s", planID, plan.Revision, stepID, digest)
|
||
step, token, err := p.store.ClaimStep(ctx, scope, planID, stepID, workerID, until, now, key)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if step == nil {
|
||
return nil
|
||
}
|
||
if obs, ok := p.owner.(FenceObserver); ok {
|
||
obs.ObserveFence(step.StepID, token)
|
||
}
|
||
out = &ClaimedStep{Plan: *plan, Step: *step, WorkerID: workerID, FencingToken: token, IdempotencyKey: step.IdempotencyKey}
|
||
return nil
|
||
})
|
||
return out, err
|
||
}
|
||
|
||
func (p *Plans) ExecuteClaimed(ctx context.Context, claimed *ClaimedStep) (*PlanView, error) {
|
||
if claimed == nil {
|
||
return nil, ErrPlanNotFoundOrHidden
|
||
}
|
||
if p.owner == nil {
|
||
return nil, ErrPlanActionForbidden
|
||
}
|
||
// FIND-FLOW-01 回归：执行器硬读 ExecutionMode——validated_only 步骤
|
||
// 一律拒绝真实执行，无论上游状态机如何推进。步骤级 mode 优先，
|
||
// 缺省回退计划级。
|
||
effectiveMode := claimed.Plan.ExecutionMode
|
||
if claimed.Step.ExecutionMode != "" {
|
||
effectiveMode = claimed.Step.ExecutionMode
|
||
}
|
||
if effectiveMode == ModeValidatedOnly {
|
||
return nil, ErrPlanModeNotExecutable
|
||
}
|
||
exec := StepExec{
|
||
Principal: claimed.Plan.Principal,
|
||
PlanID: claimed.Plan.PlanID,
|
||
StepID: claimed.Step.StepID,
|
||
ActionKey: claimed.Step.ActionKey,
|
||
IdempotencyKey: claimed.IdempotencyKey,
|
||
ActionDigest: claimed.Step.ActionDigest,
|
||
ExpectedResourceVersion: claimed.Step.ExpectedResourceVersion,
|
||
FencingToken: claimed.FencingToken,
|
||
Slots: claimed.Step.Slots,
|
||
SafeSummary: claimed.Step.SafeSummary,
|
||
}
|
||
if auth := p.stepAuthorizer(); auth != nil {
|
||
if err := auth.Authorize(ctx, exec); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
rec, err := p.owner.Execute(ctx, exec)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return p.writeReceipt(ctx, claimed, rec)
|
||
}
|
||
|
||
func (p *Plans) RecoverReceipt(ctx context.Context, claimed *ClaimedStep) (*PlanView, error) {
|
||
if claimed == nil || p.owner == nil {
|
||
return nil, ErrPlanNotFoundOrHidden
|
||
}
|
||
rec, ok, err := p.owner.LookupReceipt(ctx, claimed.IdempotencyKey)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if !ok {
|
||
return nil, ErrPlanActionVersionConflict
|
||
}
|
||
return p.writeReceipt(ctx, claimed, rec)
|
||
}
|
||
|
||
func (p *Plans) writeReceipt(ctx context.Context, claimed *ClaimedStep, rec Receipt) (*PlanView, error) {
|
||
now := p.now()
|
||
var view *PlanView
|
||
err := p.withTx(ctx, func(ctx context.Context, scope uow.TxScope) error {
|
||
ok, err := p.store.WriteStepReceipt(ctx, scope, claimed.WorkerID, claimed.FencingToken, Step{
|
||
PlanID: claimed.Plan.PlanID,
|
||
StepID: claimed.Step.StepID,
|
||
ResultRef: rec.ResultRef,
|
||
ResultDigest: rec.ResultDigest,
|
||
SafeSummary: rec.SafeSummary,
|
||
}, now)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !ok {
|
||
return ErrPlanActionVersionConflict
|
||
}
|
||
plan, err := p.store.GetPlan(ctx, scope, claimed.Plan.PlanID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
view = toView(plan)
|
||
return p.emitFact(ctx, plan, claimed.Step.StepID, now)
|
||
})
|
||
return view, err
|
||
}
|
||
|
||
func (p *Plans) ExpireDue(ctx context.Context) (int, error) {
|
||
var n int
|
||
err := p.withTx(ctx, func(ctx context.Context, scope uow.TxScope) error {
|
||
var err error
|
||
n, err = p.store.ExpireWaiting(ctx, scope, p.now())
|
||
return err
|
||
})
|
||
return n, err
|
||
}
|
||
|
||
func (p *Plans) PurgeClosed(ctx context.Context, olderThan time.Time) (int, error) {
|
||
var n int
|
||
err := p.withTx(ctx, func(ctx context.Context, scope uow.TxScope) error {
|
||
var err error
|
||
n, err = p.store.PurgeClosed(ctx, scope, olderThan)
|
||
return err
|
||
})
|
||
return n, err
|
||
}
|
||
|
||
func toView(plan *Plan) *PlanView {
|
||
if plan == nil {
|
||
return nil
|
||
}
|
||
v := &PlanView{
|
||
PlanID: plan.PlanID,
|
||
Revision: plan.Revision,
|
||
Status: plan.Status,
|
||
ExecutionMode: plan.ExecutionMode,
|
||
ExpiresAt: plan.ExpiresAt,
|
||
SafeSummary: plan.SafeSummary,
|
||
}
|
||
for _, s := range plan.Steps {
|
||
v.Steps = append(v.Steps, StepView{
|
||
StepID: s.StepID,
|
||
ActionKey: s.ActionKey,
|
||
Status: s.Status,
|
||
ExecutionMode: s.ExecutionMode,
|
||
DependsOn: slices.Clone(s.DependsOn),
|
||
SafeSummary: s.SafeSummary,
|
||
})
|
||
}
|
||
return v
|
||
}
|