feat(template): support authenticated input predicates

internal/core/application/execution/batch.go

@@ -230,7 +230,7 @@ func (b *batchWorker) intentOf(ctx context.Context, inst *delegationapp.Instance
 	}
 	in, err := execution.BuildIntent(inst.Template, inst.Delegation.Inputs, e.Coins, txs, template.Fee{Intent: info.IntentFees, Dust: info.Dust}, s.cosignerKey.PubKey())
 	if err != nil {
-		return nil, err
+		return nil, localBuildError(inst, err)
 	}
 	if in.HasArkadePacket {
 		if in.Proof, err = s.emulator.SubmitIntent(ctx, *in); err != nil {

internal/core/application/execution/eligibility_test.go

@@ -0,0 +1,168 @@
+package execution
+
+import (
+	"context"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"testing"
+	"time"
+
+	arklib "github.com/arkade-os/arkd/pkg/ark-lib"
+	delegationapp "github.com/arkade-os/executor/internal/core/application/delegation"
+	registryapp "github.com/arkade-os/executor/internal/core/application/registry"
+	"github.com/arkade-os/executor/internal/core/domain/delegation"
+	"github.com/arkade-os/executor/internal/core/domain/execution"
+	"github.com/arkade-os/executor/internal/core/domain/registry"
+	"github.com/arkade-os/executor/internal/core/ports"
+	"github.com/arkade-os/executor/pkg/template"
+	"github.com/btcsuite/btcd/wire/v2"
+	"github.com/stretchr/testify/require"
+)
+
+func TestStreamBuildFailureClassification(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		err      error
+		streamID int64
+		charge   bool
+	}{
+		{"stream ineligible", template.ErrIneligible, 1, false},
+		{"client ineligible", template.ErrIneligible, 0, true},
+		{"bad transaction", execution.ErrBadTx, 1, true},
+		{"refusal", refused{errors.New("rejected")}, 1, true},
+		{"backend error", errors.New("backend unavailable"), 1, false},
+		{"deadline", context.DeadlineExceeded, 1, false},
+		{"ineligible bad transaction", errors.Join(template.ErrIneligible, execution.ErrBadTx), 1, true},
+		{"ineligible refusal", errors.Join(template.ErrIneligible, refused{errors.New("rejected")}), 1, true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			inst := &delegationapp.Instance{Delegation: &delegation.Delegation{StreamID: tc.streamID}}
+			err := localBuildError(inst, tc.err)
+			require.EqualError(t, err, tc.err.Error())
+			require.ErrorIs(t, err, tc.err)
+			require.Equal(t, tc.charge, chargesTemplate(fmt.Errorf("worker: %w", err)))
+		})
+	}
+}
+
+func TestStreamEligibilityDoesNotDisableTemplate(t *testing.T) {
+	for _, typ := range []template.Type{template.Offchain, template.Intent} {
+		for _, streamID := range []int64{0, 1} {
+			t.Run(fmt.Sprintf("%s/stream_%d", typ, streamID), func(t *testing.T) {
+				var doc map[string]any
+				require.NoError(t, json.Unmarshal(fixture(t, "minimal.json"), &doc))
+				doc["type"] = string(typ)
+				delete(doc, "fees")
+				doc["inputs"].([]any)[0].(map[string]any)["predicate"] = "00"
+				raw, err := json.Marshal(doc)
+				require.NoError(t, err)
+				tmpl, err := template.Parse(raw, nil)
+				require.NoError(t, err)
+				inst := instance(t, raw, nil)
+				d := delegation.Delegation{ID: 1, StreamID: streamID, TemplateID: tmpl.ID(), Status: delegation.StatusActive}
+				d.Inputs = []delegation.Input{{Name: "funds", Script: hex.EncodeToString(inst.PkScript(0)), Tapscripts: inst.Tapscripts(0)}}
+				source := wire.NewMsgTx(3)
+				source.AddTxIn(&wire.TxIn{})
+				source.AddTxOut(wire.NewTxOut(10000, inst.PkScript(0)))
+				coin := delegation.Coin{Outpoint: source.TxHash().String() + ":0", DelegationID: d.ID, Script: d.Inputs[0].Script, Amount: 10000, Status: delegation.Present}
+				templates := &eligibilityTemplates{row: registry.Template{ID: tmpl.ID(), Document: raw, Status: registry.TemplateStatusActive}}
+				registryService := registryapp.NewService(nil, templates, registryapp.Limits{})
+				key := testKey(1)
+				delegations := delegationapp.NewService(registryService, &eligibilityDelegations{row: d}, nil, nil, key.PubKey(), nil, key.PubKey(), key.PubKey(), arklib.BitcoinRegTest, delegationapp.Limits{})
+				coins, attempts := &eligibilityCoins{}, &eligibilityAttempts{}
+				s := &Service{ctx: t.Context(), delegations: delegations, templates: registryService, coins: coins, attempts: attempts,
+					indexer: &eligibilityIndexer{tx: source}, ark: &eligibilitySigner{}, cosignerKey: key, info: ports.ArkInfo{Dust: 330}, limits: Limits{TemplateMaxFailures: 10}}
+				bound := &delegationapp.Instance{Delegation: &d, Template: inst, Type: typ}
+				if typ == template.Offchain {
+					_, _, err = s.build(t.Context(), bound, []delegation.Coin{coin})
+				} else {
+					_, err = (&batchWorker{s: s}).intentOf(t.Context(), bound, execution.Entry{Coins: []delegation.Coin{coin}})
+				}
+				require.ErrorIs(t, err, template.ErrIneligible)
+				require.EqualError(t, err, "build: ineligible: predicate must return one canonical true")
+				wrapped := fmt.Errorf("worker: %w", err)
+				for attempt := range 11 {
+					s.failed(attemptFailed{attempt: delegation.Attempt{DelegationID: 1, Coins: []string{coin.Outpoint}, Status: delegation.Submitted}, err: wrapped, retryIn: 5 * time.Second}, now.Add(time.Duration(attempt)*time.Minute))
+				}
+				require.Len(t, attempts.rows, 11)
+				require.Len(t, coins.ready, 11)
+				for attempt, at := range coins.ready {
+					require.Equal(t, now.Add(time.Duration(attempt)*time.Minute+5*time.Second), at)
+					require.Equal(t, delegation.Failed, attempts.rows[attempt].Status)
+					require.Equal(t, []string{coin.Outpoint}, attempts.rows[attempt].Coins)
+				}
+				if streamID != 0 {
+					require.Zero(t, templates.row.Failures)
+					require.Zero(t, s.templateFailures)
+					_, err := registryService.Usable(t.Context(), tmpl.ID())
+					require.NoError(t, err)
+				} else {
+					require.Equal(t, 11, templates.row.Failures)
+					require.EqualValues(t, 11, s.templateFailures)
+					require.Equal(t, registry.TemplateStatusDisabled, templates.row.Status)
+				}
+			})
+		}
+	}
+}
+
+type eligibilityIndexer struct {
+	ports.Indexer
+	tx *wire.MsgTx
+}
+
+func (r *eligibilityIndexer) VirtualTxs(context.Context, []string) (map[string]*wire.MsgTx, error) {
+	return map[string]*wire.MsgTx{r.tx.TxHash().String(): r.tx}, nil
+}
+
+type eligibilitySigner struct{ ports.Signer }
+
+func (*eligibilitySigner) Info(context.Context) (ports.ArkInfo, error) {
+	return ports.ArkInfo{Dust: 330}, nil
+}
+
+type eligibilityTemplates struct {
+	registry.TemplateRepository
+	row registry.Template
+}
+
+func (r *eligibilityTemplates) Get(context.Context, string) (*registry.Template, error) {
+	row := r.row
+	return &row, nil
+}
+func (r *eligibilityTemplates) AddOrUpdate(_ context.Context, row *registry.Template) error {
+	r.row = *row
+	return nil
+}
+
+type eligibilityDelegations struct {
+	delegation.DelegationRepository
+	row delegation.Delegation
+}
+
+func (r *eligibilityDelegations) Get(context.Context, int64) (*delegation.Delegation, error) {
+	row := r.row
+	return &row, nil
+}
+
+type eligibilityAttempts struct {
+	delegation.AttemptRepository
+	rows []delegation.Attempt
+}
+
+func (r *eligibilityAttempts) AddOrUpdate(_ context.Context, row *delegation.Attempt) error {
+	r.rows = append(r.rows, *row)
+	return nil
+}
+
+type eligibilityCoins struct {
+	delegation.CoinRepository
+	ready []time.Time
+}
+
+func (r *eligibilityCoins) Release(_ context.Context, _ []string, ready time.Time) error {
+	r.ready = append(r.ready, ready)
+	return nil
+}

internal/core/application/execution/offchain.go

@@ -84,7 +84,7 @@ func (s *Service) build(ctx context.Context, inst *delegationapp.Instance, coins
 	fee := template.Fee{Intent: s.info.IntentFees, Dust: s.info.Dust}
 	built, err := execution.BuildOffchain(inst.Template, inst.Delegation.Inputs, coins, txs, fee, s.info.CheckpointTapscript)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, localBuildError(inst, err)
 	}
 	sources := make([]*wire.MsgTx, len(coins))
 	for i, c := range coins {
@@ -157,11 +157,25 @@ func (r refused) Unwrap() error { return r.error }
 
 func (refused) Is(target error) bool { return target == errRefused }
 
+type streamIneligible struct{ error }
+
+func (e streamIneligible) Unwrap() error { return e.error }
+
+func localBuildError(inst *delegationapp.Instance, err error) error {
+	if inst.Delegation.StreamID != 0 && errors.Is(err, template.ErrIneligible) {
+		return streamIneligible{err}
+	}
+	return err
+}
+
 // chargesTemplate says a failed attempt counts against its template: the template could not build
 // it, or arkd or the emulator refused it. Outages, the daemon's own shutdown and sources fees or dust refuse do not.
 func chargesTemplate(err error) bool {
 	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, template.ErrUnaffordable) {
 		return false
 	}
-	return errors.Is(err, template.ErrIneligible) || errors.Is(err, execution.ErrBadTx) || errors.Is(err, errRefused)
+	// Stream inputs are untrusted; local ineligibility does not indict their shared template.
+	var streamed streamIneligible
+	return errors.Is(err, execution.ErrBadTx) || errors.Is(err, errRefused) ||
+		errors.Is(err, template.ErrIneligible) && !errors.As(err, &streamed)
 }

pkg/template/build.go

@@ -56,6 +56,9 @@ func (i *Instance) Build(sources []*Source, fee Fee) (*Action, error) {
 	if err != nil {
 		return nil, err
 	}
+	if err := i.predicates(d); err != nil {
+		return nil, err
+	}
 
 	charged := uint64(0)
 	if t.typ == Intent {

pkg/template/draft.go

@@ -32,10 +32,11 @@ type Source struct {
 
 // draft is the transaction expressions see: inputs at their final position, no output; not safe for concurrent use.
 type draft struct {
-	tx      *wire.MsgTx
-	sources []*Source         // aligned with tx.TxIn; nil for an intent's message input
-	bound   []bool            // aligned with tx.TxIn; false for a source whose Tx is not its outpoint's
-	secrets map[string][]byte // plaintexts opened for this draft
+	tx              *wire.MsgTx
+	sources         []*Source         // aligned with tx.TxIn; nil for an intent's message input
+	bound           []bool            // aligned with tx.TxIn; false for a source whose Tx is not its outpoint's
+	secrets         map[string][]byte // plaintexts opened for this draft
+	predicateBudget *predicateBudget
 }
 
 func newDraft(typ Type, sources []*Source) *draft {
@@ -65,6 +66,12 @@ func (d *draft) index(slot int) int {
 }
 
 func (d *draft) eval(tokens []string, input int, lookup asm.Lookup) ([]byte, error) {
+	// Assembly may resolve secrets whose ciphertext bindings evaluate further expressions.
+	if d.predicateBudget != nil {
+		if err := d.predicateBudget.expression(tokens); err != nil {
+			return nil, err
+		}
+	}
 	script, err := asm.Assemble(tokens, lookup)
 	if err != nil {
 		return nil, fmt.Errorf("%w: expression: %v", ErrIneligible, err)

pkg/template/fixtures_test.go

@@ -34,7 +34,7 @@ func TestFixtures(t *testing.T) {
 	var ids map[string]string
 	require.NoError(t, json.Unmarshal(fixture(t, "ids.json"), &ids))
 	docs := artifacts(t)
-	for _, name := range []string{"renew.json", "board.json", "counter.json", "vhtlc_claim.json", "vhtlc_refund.json", "claim.json", "offline_receive.json", "onchain_release.json", "counter_seed.json", "minimal.json"} {
+	for _, name := range []string{"renew.json", "board.json", "counter.json", "vhtlc_claim.json", "vhtlc_refund.json", "claim.json", "offline_receive.json", "offline_receive_predicate.json", "onchain_release.json", "counter_seed.json", "minimal.json"} {
 		t.Run(name, func(t *testing.T) {
 			tmpl, err := Parse(fixture(t, name), docs)
 			require.NoError(t, err)

pkg/template/input.go

@@ -31,6 +31,7 @@ type input struct {
 	function, leaf string
 	arguments      []byteTemplate          // covenant inputs, ABI order
 	witness        map[string]byteTemplate // non-signature witness items
+	predicate      *byteTemplate
 }
 
 func (c *compiler) input(s *schema.Input) error {
@@ -55,6 +56,13 @@ func (c *compiler) input(s *schema.Input) error {
 	if err := c.spend(&in, &s.Spend); err != nil {
 		return err
 	}
+	if s.Predicate != nil {
+		b, err := c.bind(*s.Predicate, "bytes", true)
+		if err != nil {
+			return err
+		}
+		in.predicate = &b
+	}
 	c.t.inputs = append(c.t.inputs, in)
 	return nil
 }

pkg/template/internal/schema/schema.go

@@ -52,11 +52,12 @@ type Secret struct {
 }
 
 type Input struct {
-	Name     string       `json:"name"`                            // identifier
-	Type     string       `json:"type,omitempty" schema:"opt"`     // offchain or onchain; defaults to the template's kind
-	Schedule *Schedule    `json:"schedule,omitempty" schema:"opt"` // defaults to immediate
-	Contract Construction `json:"contract"`
-	Spend    Spend        `json:"spend"`
+	Name      string       `json:"name"`                            // identifier
+	Type      string       `json:"type,omitempty" schema:"opt"`     // offchain or onchain; defaults to the template's kind
+	Schedule  *Schedule    `json:"schedule,omitempty" schema:"opt"` // defaults to immediate
+	Contract  Construction `json:"contract"`
+	Spend     Spend        `json:"spend"`
+	Predicate *Binding     `json:"predicate,omitempty" schema:"opt"`
 }
 
 // Schedule is when an input becomes eligible; exactly one field is set.