Make fresh Flow runs self-initializing

boatstack/cmd/boatstack-helper/control_bundle.go

@@ -303,6 +303,7 @@ func bindTrustedRequestControlBundle(ctx context.Context, request *surfaces.Requ
 	}
 	request.ControlBundle = nil
 	request.ControlBundleFingerprint = ""
+	request.ControlBundleRevision = ""
 	if request.ProgramID == "" && !controlBundleRequired(request.TransitionID) {
 		return nil
 	}
@@ -315,12 +316,31 @@ func bindTrustedRequestControlBundle(ctx context.Context, request *surfaces.Requ
 	return nil
 }
 
-func verifyTrustedRequestControlBundle(request surfaces.Request) error {
+func verifyTrustedRequestControlBundle(ctx context.Context, request surfaces.Request) error {
 	if request.ControlBundle == nil {
 		if controlBundleRequired(request.TransitionID) {
 			return fmt.Errorf("CONTROL_BUNDLE_REQUIRED: transition %q has no trusted bundle", request.TransitionID)
 		}
 		return nil
 	}
+	if request.ControlBundleRevision != "" {
+		revision, err := boatstackruntime.ResolveCommitRevision(ctx, request.Repository, "HEAD")
+		if err != nil {
+			return err
+		}
+		if revision != request.ControlBundleRevision {
+			return &flowCommitRequiredError{
+				programID: request.ProgramID, entryID: request.EntryID, runID: request.FlowID,
+				revision: revision, controlBundleFingerprint: request.ControlBundleFingerprint, operation: request.Operation,
+				cause: fmt.Errorf("CONTROL_BUNDLE_REVISION_DRIFT: expected revision %s", request.ControlBundleRevision),
+			}
+		}
+		if err := boatstackruntime.VerifyControlBundleRevision(ctx, request.Repository, revision, request.ControlBundle.Source); err != nil {
+			return &flowCommitRequiredError{
+				programID: request.ProgramID, entryID: request.EntryID, runID: request.FlowID,
+				revision: revision, controlBundleFingerprint: request.ControlBundleFingerprint, operation: request.Operation, cause: err,
+			}
+		}
+	}
 	return boatstackruntime.VerifyControlBundleRoot(request.Repository, request.ControlBundle.Source)
 }

boatstack/cmd/boatstack-helper/delegation_command.go

@@ -63,8 +63,15 @@ func runFlowAuthorize(arguments []string) error {
 	if expiresIn < 0 {
 		return fmt.Errorf("flow authorize --expires-in cannot be negative")
 	}
+	// Authorization reconstructs the exact request surfaced after candidate
+	// selection. Ordinary unbound resolution must not create product delegation
+	// before installation has established the control bundle.
+	options.delegationRequestProjection = true
 	bound, err := bindFlowEntry(context.Background(), options)
 	if err != nil {
+		if suspended, ok := flowCommitRequiredResponse(err, surfaces.OperationResolve); ok {
+			return renderResponse(suspended, "json")
+		}
 		return err
 	}
 	if bound.delegationRequestFingerprint == "" || requestFingerprint != bound.delegationRequestFingerprint || bound.runID != options.runID {
@@ -251,6 +258,10 @@ func runFlowContinuation(arguments []string) error {
 	if err != nil {
 		return err
 	}
+	return runFlowContinuationOptions(options)
+}
+
+func runFlowContinuationOptions(options commandOptions) error {
 	if options.programID == "" || options.entryID == "" {
 		return fmt.Errorf("flow run requires --flow and --entry")
 	}
@@ -261,9 +272,13 @@ func runFlowContinuation(arguments []string) error {
 		return fmt.Errorf("FLOW_INPUT_INVALID: --input is available only to declarative Flow entries")
 	}
 	var response surfaces.Response
+	var err error
 	for step := 0; step < 256; step++ {
 		response, err = executeContinuationStep(context.Background(), options)
 		if err != nil {
+			if suspended, ok := flowCommitRequiredResponse(err, surfaces.OperationResolve); ok {
+				return renderResponse(suspended, options.format)
+			}
 			return err
 		}
 		if response.RunID != "" {
@@ -278,7 +293,7 @@ func runFlowContinuation(arguments []string) error {
 		if err := advanceContinuation(&options, response); err != nil {
 			return err
 		}
-		if response.Delegation != nil || response.Prescription == nil || response.Receipt == nil || (response.Decision != nil && response.Decision.Kind == supervisor.DecisionTerminal) {
+		if response.Delegation != nil || response.CommitRequired != nil || response.Prescription == nil || response.Receipt == nil || (response.Decision != nil && response.Decision.Kind == supervisor.DecisionTerminal) {
 			return renderResponse(response, options.format)
 		}
 	}
@@ -316,7 +331,7 @@ func executeContinuationStep(ctx context.Context, options commandOptions) (surfa
 	if err != nil {
 		return surfaces.Response{}, err
 	}
-	if err := verifyTrustedRequestControlBundle(resolveRequest); err != nil {
+	if err := verifyTrustedRequestControlBundle(ctx, resolveRequest); err != nil {
 		resolveLease.Release()
 		return surfaces.Response{}, err
 	}
@@ -371,7 +386,7 @@ func executeContinuationStep(ctx context.Context, options commandOptions) (surfa
 		if err != nil {
 			return surfaces.Response{}, err
 		}
-		if err := verifyTrustedRequestControlBundle(resolveRequest); err != nil {
+		if err := verifyTrustedRequestControlBundle(ctx, resolveRequest); err != nil {
 			resolveLease.Release()
 			return surfaces.Response{}, err
 		}
@@ -414,7 +429,7 @@ func executeContinuationStep(ctx context.Context, options commandOptions) (surfa
 	if delegationResponse != nil {
 		return *delegationResponse, nil
 	}
-	if err := verifyTrustedRequestControlBundle(applyRequest); err != nil {
+	if err := verifyTrustedRequestControlBundle(ctx, applyRequest); err != nil {
 		return surfaces.Response{}, err
 	}
 	applied, err := kernel.Handle(ctx, applyRequest)
@@ -462,7 +477,7 @@ func stabilizeRepositoryPrescription(ctx context.Context, request surfaces.Reque
 		return surfaces.Request{}, surfaces.Response{}, true, err
 	}
 	defer lease.Release()
-	if err := verifyTrustedRequestControlBundle(rebound); err != nil {
+	if err := verifyTrustedRequestControlBundle(ctx, rebound); err != nil {
 		return surfaces.Request{}, surfaces.Response{}, true, err
 	}
 	kernel, err := standardKernel(ctx, rebound)
@@ -528,7 +543,7 @@ func bindContinuationCandidate(ctx context.Context, bound commandOptions, respon
 	if err != nil {
 		return commandOptions{}, false, err
 	}
-	if rebound.inputRequest == nil && len(rebound.parameters) <= len(bound.parameters) {
+	if rebound.inputRequest == nil && len(rebound.parameters) <= len(bound.parameters) && rebound.delegationRequestFingerprint == "" {
 		return bound, false, nil
 	}
 	return rebound, true, nil
@@ -563,5 +578,6 @@ func advanceContinuation(options *commandOptions, response surfaces.Response) er
 	options.trustedAuthorityReceipts = nil
 	options.invocationEvidence = nil
 	options.inputRequest = nil
+	options.controlBundleRevision = ""
 	return nil
 }

boatstack/cmd/boatstack-helper/delegation_runtime.go

@@ -161,6 +161,7 @@ func preflightDelegatedProgramChange(ctx context.Context, request surfaces.Reque
 		return nil, nil
 	}
 	probe := request
+	requestedOperation := request.Operation
 	probe.Operation = surfaces.OperationExplain
 	probe.Prescription = protocol.Prescription{}
 	probe.IdempotencyKey = ""
@@ -171,8 +172,8 @@ func preflightDelegatedProgramChange(ctx context.Context, request surfaces.Reque
 		return nil, err
 	}
 	defer lease.Release()
-	if err := verifyTrustedRequestControlBundle(probe); err != nil {
-		return nil, err
+	if err := verifyTrustedRequestControlBundle(ctx, probe); err != nil {
+		return nil, bindFlowCommitRequiredOperation(err, requestedOperation)
 	}
 	kernel, err := standardKernel(ctx, probe)
 	if err != nil {

boatstack/cmd/boatstack-helper/flow_command.go

@@ -21,7 +21,7 @@ import (
 	boatstackruntime "github.com/operatorstack/boatstack/boatstack/internal/runtime"
 )
 
-const flowCompilerVersion = "control-program.compiler.4"
+const flowCompilerVersion = "control-program.compiler.5"
 
 type flowCommandOptions struct {
 	repository string