weichen
c07d4020d0
Persist validated_only shadow plans without executing, gate elevated tiers behind INTENT_PLAN_PROJECT_ALLOWLIST, and keep legacy chat routing for shadow.
710 lines
18 KiB
Go
710 lines
18 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
|
|
now func() time.Time
|
|
newID func(prefix string) string
|
|
lease time.Duration
|
|
ttl time.Duration
|
|
}
|
|
|
|
type Options struct {
|
|
Owner ActionOwner
|
|
Authorizer StepAuthorizer
|
|
Events EventSink
|
|
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, 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 {
|
|
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,
|
|
}); 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
|
|
}
|
|
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
|
|
}
|