johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package requester

import (
	"cmp"
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"fmt"
	"log/slog"
	"slices"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/market"
	contractmarket "github.com/johnandersen777/socialweb-computer-kcp/internal/contract/market"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/guestseed"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/conds"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/netretry"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/market/participant"
)

type contractRunner struct {
	q *Requester

	name string

	mu sync.Mutex

	contract v1alpha1.ComputeContract

	state *v1alpha1.NodeClaimContract

	bids []v1alpha1.ContractBid

	chosen *v1alpha1.ContractChosen

	release *v1alpha1.ComputeContractRelease

	resumed bool

	releaseCh chan struct{}

	releaseOnce sync.Once

	done chan struct{}

	cancel context.CancelFunc

	gone atomic.Bool

	terminationNotReported atomic.Bool

	nodeJoin *v1alpha1.ComputeContractNodeJoin

	grant *v1alpha1.NodeClaimGrant

	secretsURL string

	secretsFetchedAt string

	secrets *secretsState

	rbacRKey string

	atircRevoke sync.Once

	releaseRequested string

	association *contractAssociation

	associationMu sync.Mutex

	associationKick chan struct{}

	bundleMu *sync.Mutex

	associationRecords map[string]string

	associationAttempts int

	associationRetry *time.Timer

	atircMint *atircMintOutcome
}

func newContractRunner(q *Requester, contract v1alpha1.ComputeContract) *contractRunner {
	r := &contractRunner{
		q:         q,
		name:      contract.Name,
		contract:  contract,
		bids:      append([]v1alpha1.ContractBid(nil), contract.Status.Bids...),
		releaseCh: make(chan struct{}),
		done:      make(chan struct{}),

		associationKick: make(chan struct{}, 1),

		bundleMu: q.bundleLock(contract.Name),
	}
	if contract.Status.Chosen != nil {
		chosen := *contract.Status.Chosen
		r.chosen = &chosen
	}
	if contract.Status.Release != nil {
		release := *contract.Status.Release
		r.release = &release
	}
	if contract.Status.Contract != nil {
		r.state = cloneNodeClaimContract(contract.Status.Contract)
		r.resumed = recordedStage(contract.Status.Contract)
	}
	r.nodeJoin = contract.Spec.NodeJoin
	r.secretsURL = contract.Status.SecretsURL
	r.secretsFetchedAt = contract.Status.SecretsFetchedAt
	if contract.Status.Grant != nil {
		grant := *contract.Status.Grant
		r.grant = &grant
	}
	return r
}

func recordedStage(state *v1alpha1.NodeClaimContract) bool {
	if state == nil {
		return false
	}
	return state.VMURI != "" || state.RFPURI != "" || state.AcceptURI != "" || state.ReceiptURI != "" || state.Release != nil
}

func (r *contractRunner) Update(contract v1alpha1.ComputeContract) {
	r.mu.Lock()
	r.contract = contract
	r.mu.Unlock()
	if contract.DeletionTimestamp != nil || contract.Spec.Release {
		r.requestRelease()
		return
	}
	r.associateLater(contract)
}

func (r *contractRunner) releaseStatus() *v1alpha1.ComputeContractRelease {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.release == nil {
		return nil
	}
	out := *r.release
	return &out
}

func (r *contractRunner) setRelease(release *v1alpha1.ComputeContractRelease) {
	r.mu.Lock()
	defer r.mu.Unlock()
	if release == nil {
		r.release = nil
		return
	}
	out := *release
	r.release = &out
}

func (r *contractRunner) requestRelease() {
	r.releaseOnce.Do(func() { close(r.releaseCh) })
}

func (r *contractRunner) untilRelease(ctx context.Context, ref func(context.Context) (string, error)) (string, error) {
	waitCtx, cancel := context.WithCancel(ctx)
	defer cancel()
	go func() {
		select {
		case <-r.releaseCh:
			cancel()
		case <-waitCtx.Done():
		}
	}()
	return ref(waitCtx)
}

func (r *contractRunner) stop() {
	r.gone.Store(true)
	if r.cancel != nil {
		r.cancel()
	}
}

func (r *contractRunner) isGone() bool {
	return r.gone.Load()
}

func (r *contractRunner) released() bool {
	select {
	case <-r.releaseCh:
		return true
	default:
		return false
	}
}

func (r *contractRunner) snapshot() v1alpha1.ComputeContract {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.contract
}

func (r *contractRunner) stateCopy() *v1alpha1.NodeClaimContract {
	r.mu.Lock()
	defer r.mu.Unlock()
	return cloneNodeClaimContract(r.state)
}

func (r *contractRunner) setState(state *v1alpha1.NodeClaimContract) {
	r.mu.Lock()
	r.state = state
	r.mu.Unlock()
}

func (r *contractRunner) log() *slog.Logger {
	if r.isGone() {
		return slog.New(slog.DiscardHandler)
	}
	return r.q.log.With("contract", r.name)
}

func (r *contractRunner) Run(ctx context.Context) {
	defer close(r.done)
	defer r.cleanup()
	contract := r.snapshot()
	if contract.Spec.Adopt {
		r.runAdopt(ctx, contract)
		return
	}
	state := r.stateCopy()
	tokenPathConflict := CheckTokenPaths(contract, nil)
	if tokenPathConflict != nil {
		if (state == nil || state.RFPURI == "") && !r.released() {
			r.log().Warn("requester_contract_refused", "err", tokenPathConflict.Error())
			r.writeStatus(ctx, v1alpha1.ComputeContractFailed, "refused: "+tokenPathConflict.Error(), state, nil, nil, "refused", 1)
			return
		}
		r.log().Warn("requester_contract_token_path_conflict", "err", tokenPathConflict.Error())
	}
	if contract.Status.Phase == v1alpha1.ComputeContractReleased && state != nil && state.Receipt != nil {
		r.resumeReleased(ctx, state)
		return
	}
	if contract.Spec.Agent != nil && (state == nil || state.RFPURI == "") && !r.released() {
		if err := r.ensureAgentBundle(ctx, contract); err != nil {
			r.recordFailure(ctx, state, err, 1)
			return
		}
		if err := r.q.checkAgentHandle(ctx); err != nil {
			r.log().Warn("requester_contract_refused", "err", err.Error())
			r.writeStatus(ctx, v1alpha1.ComputeContractFailed, "refused: "+err.Error(), state, nil, nil, "refused", 1)
			return
		}
	}
	if contract.Spec.SecretsRef != nil && contract.Status.SecretsFetchedAt == "" {
		if err := r.untilSettled(ctx, "ensureSecrets", func() error { return r.ensureSecrets(ctx, contract) }); err != nil {
			if !r.q.stopping() {
				r.recordFailure(ctx, r.stateCopy(), err, 1)
			}
			return
		}
	}
	state = r.stateCopy()
	if state != nil && state.Receipt != nil {
		if err := r.resumeReceipt(ctx, state); err != nil && !r.q.stopping() {
			r.recordFailure(ctx, r.stateCopy(), err, 1)
		}
		return
	}
	if r.released() {
		if r.recoverReceipt(ctx, state) {
			if err := r.releaseContract(ctx, state, winnerDID(state)); err != nil && !r.q.stopping() {
				r.recordFailure(ctx, r.stateCopy(), err, 1)
			}
			return
		}
		if acceptedWithoutReceipt(state) {
			if err := r.releaseDeniedAccept(ctx, state, winnerDID(state)); err != nil && !r.q.stopping() {
				r.recordFailure(ctx, r.stateCopy(), err, 1)
			}
			return
		}
		r.writeStatus(ctx, v1alpha1.ComputeContractReleased, "released before a VM was held", state, r.currentBids(), r.currentChosen(), "released", 1)
		return
	}
	err := r.requestOnce(ctx)
	if err == nil {
		return
	}
	if r.q.stopping() {
		r.log().Info("requester_contract_left_for_restart", "err", err)
		return
	}
	if r.released() {
		st := r.stateCopy()
		if st != nil && st.Receipt == nil && r.recoverReceipt(ctx, st) {
			if err := r.releaseContract(ctx, st, winnerDID(st)); err != nil && !r.q.stopping() {
				r.recordFailure(ctx, r.stateCopy(), err, 1)
			}
			return
		}
		if st == nil || st.Receipt == nil {
			r.writeStatus(ctx, v1alpha1.ComputeContractReleased, "released before a VM was held", st, r.currentBids(), r.currentChosen(), "released", 1)
			return
		}
	}
	r.recordFailure(ctx, r.stateCopy(), err, 1)
}

func (r *contractRunner) recordFailure(ctx context.Context, state *v1alpha1.NodeClaimContract, err error, attempts int32) {
	if r.isGone() {
		return
	}
	if state != nil && state.Release != nil && !state.Release.OK {
		r.writeStatus(ctx, v1alpha1.ComputeContractReleasing, "the release did not finish: "+err.Error(), state, r.currentBids(), r.currentChosen(), "releaseFailed", attempts)
		return
	}
	if r.released() && acceptedWithoutReceipt(state) {
		r.log().Warn("requester_contract_release_failed", "err", err)
		r.writeStatus(ctx, v1alpha1.ComputeContractFailed, "the release did not find the receipt of the accept: "+err.Error(), state, r.currentBids(), r.currentChosen(), "releaseFailed", attempts)
		return
	}
	if r.released() && (state == nil || state.Receipt == nil) {
		r.writeStatus(ctx, v1alpha1.ComputeContractReleased, "released before a VM was held", state, r.currentBids(), r.currentChosen(), "released", attempts)
		return
	}
	r.log().Warn("requester_contract_failed", "err", err)
	r.writeStatus(ctx, v1alpha1.ComputeContractFailed, "the contract flow failed: "+err.Error(), state, r.currentBids(), r.currentChosen(), "failed", attempts)
	if state != nil && state.Receipt != nil && state.Termination == nil {
		if waitErr := r.waitForRelease(ctx, state, state.Receipt.URI, winnerDID(state)); waitErr != nil {
			r.log().Info("requester_contract_failed_hold_ended", "err", waitErr)
		}
	}
}

func (q *Requester) generatedVMName() string {
	raw := make([]byte, 4)
	if _, err := rand.Read(raw); err != nil {
		return fmt.Sprintf("%s%d", DefaultVMNamePrefix, time.Now().UnixNano()%100000000)
	}
	return DefaultVMNamePrefix + hex.EncodeToString(raw)
}

func (r *contractRunner) requestOnce(ctx context.Context) error {
	q := r.q
	contract := r.snapshot()
	vmName := strings.TrimSpace(contract.Spec.VMName)
	if vmName == "" {
		vmName = q.generatedVMName()
	}
	authorizedKeys := append([]string(nil), contract.Spec.SshAuthorizedKeys...)
	authorizedKeys = append(authorizedKeys, q.cfg.GuestAuthorizedKeys...)
	userData := contract.Spec.UserData
	if userData == "" {
		userData = defaultUserData(authorizedKeys)
	}
	var err error
	userData, err = guestseed.WithIrohTransport(userData, authorizedKeys, nil)
	if err != nil {
		return fmt.Errorf("the guest iroh transport: %w", err)
	}
	userData = withGuestCACerts(userData, q.guestCAPEM())
	userData, err = r.contractModules(userData, contract)
	if err != nil {
		return err
	}
	policyName := cmp.Or(strings.TrimSpace(contract.Spec.Policy), q.cfg.PolicyWorkflow, DefaultPolicyWorkflow)
	policyArgs, err := participant.ParsePolicyArgs(contract.Spec.PolicyArgs)
	if err != nil {
		return err
	}
	state := r.stateCopy()
	if state == nil {
		state = &v1alpha1.NodeClaimContract{
			Version:    new(1),
			Phase:      v1alpha1.ContractRequested,
			Timestamps: &v1alpha1.ContractTimestamps{Requested: q.nowISO()},
		}
	}
	state.VMName = cmp.Or(state.VMName, vmName)
	state.RequesterDid = cmp.Or(state.RequesterDid, q.callbackDID())
	state.MarketDid = cmp.Or(state.MarketDid, q.marketDID)
	if state.Version == nil {
		state.Version = new(1)
	}
	if state.Timestamps == nil {
		state.Timestamps = &v1alpha1.ContractTimestamps{Requested: q.nowISO()}
	}
	state.Phase = cmp.Or(state.Phase, v1alpha1.ContractRequested)
	r.setState(state)
	r.step(ctx, "preparing", 1)

	if state.Payload == nil || state.Payload.URI == "" {
		payloadRecord, err := market.EncodeRecord(market.ComputeVM{Role: vmName, Kind: string(contract.Spec.ComputeKind),
			Cpus: DefaultComputeCPUs, Mem: DefaultComputeMem, Disk: DefaultComputeDisk, UserData: userData, CreatedAt: q.nowISO()})
		if err != nil {
			return fmt.Errorf("encode the compute.vm record: %w", err)
		}
		payload, err := q.repo.CreateRecord(ctx, q.repoDID, market.ComputeVMNSID, payloadRecord)
		if err != nil {
			return fmt.Errorf("create the compute.vm record: %w", err)
		}
		state.VMURI = payload.URI
		state.Payload = &commonv1.StrongRef{URI: payload.URI, CID: payload.CID}
		state.Phase = v1alpha1.ContractRFPSubmitted
	} else {
		state.VMURI = state.Payload.URI
	}

	r.step(ctx, "creatingRfp", 1)
	var policyRef *market.StrongRef
	policyWorkflow := PolicyWorkflowName(policyName)
	workflowText, ok := PolicyWorkflowText(policyName)
	if !ok {
		return fmt.Errorf("policy %q names no bundled workflow", policyName)
	}
	if state.Policy != nil && state.Policy.URI != "" {
		policyRef = &market.StrongRef{Type: market.StrongRefNSID, URI: state.Policy.URI, CID: state.Policy.CID}
	} else {
		policyRecord := BuildPolicyRecord(policyName, policyWorkflow, workflowText, q.marketDID, policyArgs, q.nowISO())
		created, err := q.repo.CreateRecord(ctx, q.repoDID, PolicyGhaLiteNSID, policyRecord)
		if err != nil {
			return fmt.Errorf("create the policy record: %w", err)
		}
		policyRef = &created
		state.Policy = &commonv1.StrongRef{URI: created.URI, CID: created.CID}
	}
	gate, err := q.attachedPolicy(ctx, *policyRef, policyWorkflow)
	if err != nil {
		return err
	}
	rfpURI, rfpCID := "", ""
	if state.RFP != nil && state.RFP.URI != "" {
		rfpURI, rfpCID = state.RFP.URI, state.RFP.CID
	} else {
		submitBid, err := r.untilRelease(ctx, q.submitBidRef)
		if err != nil {
			return err
		}
		rfpOptions := market.RFPOptions{
			Payload:   market.RecordRef{URI: state.Payload.URI, CID: state.Payload.CID},
			SubmitBid: submitBid,
			Ingress:   q.ingressRecords(ctx),
			Domain:    "compute",
			CreatedAt: q.nowISO(),
		}
		rfpOptions.Policies = []market.RecordRef{{URI: policyRef.URI, CID: policyRef.CID}}
		if q.cfg.Spec.SuspendParticipation {
			return ErrParticipationSuspended
		}
		rfp, err := market.CreateRFP(ctx, q.repo, q.repoDID, rfpOptions, q.signer)
		if err != nil {
			return fmt.Errorf("create the market.rfp record: %w", err)
		}
		state.RFPURI = rfp.URI
		state.RFP = &commonv1.StrongRef{URI: rfp.URI, CID: rfp.CID}
		state.Timestamps.RFPSubmitted = q.nowISO()
		rfpURI, rfpCID = rfp.URI, rfp.CID
	}
	r.writeStatus(ctx, v1alpha1.ComputeContractRequesting, "the RFP is submitted; collecting bids", state, nil, nil, "collectingBids", 1)

	if state.Accept != nil && state.Accept.URI != "" {
		return r.resumeAccepted(ctx, state)
	}

	spec := r.snapshot().Spec
	only := BidderDIDList(spec.BidderDids)
	discovery, err := q.market.Discover(ctx, q.now(), only)
	if err != nil {
		r.log().Warn("requester_bidder_discovery_failed", "err", err)
	}
	discovery = discovery.Without(spec.DenyBidderDids)
	bidders := discovery.Bidders
	noBidders := ""
	if len(bidders) == 0 {
		noBidders = "requester_no_bidders: " + discovery.Why()
		r.log().Warn("requester_no_bidders", "restricted", discovery.Restricted, "candidates", discovery.Candidates, "dropped", discovery.DroppedText())
		if discovery.Restricted {
			state.Outcome = v1alpha1.ContractOutcomeNoBids
			state.Error = noBidders
			return errors.New(state.Error)
		}
		r.writeStatus(ctx, v1alpha1.ComputeContractRequesting, "the RFP is submitted; "+noBidders+"; waiting for bids", state, nil, nil, "collectingBids", 1)
	}
	q.rememberBidders(bidders)
	considered := consideredBidderDIDs(bidders, only)
	catchUpCtx, stopCatchUp := context.WithCancel(ctx)
	r.catchUpBids(catchUpCtx, considered, rfpURI, rfpCID)
	r.solicitBids(ctx, bidders, rfpURI, rfpCID)

	winner, allBids, err := r.collectBids(ctx, rfpURI, gate, only)
	stopCatchUp()
	if err != nil {
		return err
	}
	if r.released() {
		r.writeStatus(ctx, v1alpha1.ComputeContractReleased, "released before a VM was held", state, allBids, nil, "released", 1)
		return nil
	}
	r.mu.Lock()
	r.bids = allBids
	r.mu.Unlock()
	if winner == nil {
		if len(allBids) == 0 {
			state.Outcome = v1alpha1.ContractOutcomeNoBids
			state.Error = fmt.Sprintf("no bids received within %s", q.cfg.BidWindow)
			if noBidders != "" {
				state.Error += "; " + noBidders
			}
			return errors.New(state.Error)
		}
		state.Outcome = v1alpha1.ContractOutcomeBidsDenied
		state.Error = fmt.Sprintf("all %d bids were denied by policy", len(allBids))
		return errors.New(state.Error)
	}
	if r.released() {
		r.writeStatus(ctx, v1alpha1.ComputeContractReleased, "released before a VM was held", state, allBids, nil, "released", 1)
		return nil
	}

	bidderByDID := map[string]Bidder{}
	for _, bidder := range bidders {
		if _, ok := bidderByDID[bidder.DID]; !ok {
			bidderByDID[bidder.DID] = bidder
		}
	}
	winnerBidder, ok := bidderByDID[winner.DID]
	if !ok {
		winnerBidder, ok = q.bidderForBid(ctx, *winner)
	}
	if !ok {
		return fmt.Errorf("no endpoint is known for the winning bidder %s", winner.DID)
	}
	chosen := &v1alpha1.ContractChosen{
		URI:        winner.URI,
		Did:        winner.DID,
		Reason:     v1alpha1.ContractChosenBestWithinWindow,
		Considered: len(allBids),
	}
	decision := q.policy.Evaluate(ctx, gate.Input(PolicyInput{
		SelfDid:          q.marketDID,
		SubjectDid:       winner.DID,
		CounterpartyDid:  winner.DID,
		RootRequesterDid: q.marketDID,
		RFP:              &PolicyRef{URI: rfpURI, CID: rfpCID},
		Bid:              &PolicyRef{URI: winner.URI, CID: winner.CID},
	}))
	if !decision.Allow {
		state.Outcome = v1alpha1.ContractOutcomePolicyRejected
		state.Error = "winner rejected by policy: " + decision.Reason
		return errors.New(state.Error)
	}
	chosen.Reason = v1alpha1.ContractChosenPolicy
	if winner.Free {
		chosen.Reason = v1alpha1.ContractChosenFirstFree
	}
	r.mu.Lock()
	r.chosen = chosen
	r.mu.Unlock()
	acceptRKey := acceptRKeyFor(q.repoDID, r.name, rfpURI, rfpCID, winner.URI, winner.CID)
	if err := r.prepareWorkloadGrant(ctx, state, winner, acceptRKey); err != nil {
		return err
	}
	state.Phase = v1alpha1.ContractWinnerSelected
	state.Bid = &commonv1.StrongRef{URI: winner.URI, CID: winner.CID}
	state.Winner = &v1alpha1.ContractWinner{DID: winner.DID, SubmitAccept: strings.TrimSpace(stringOf(winner.Record["submitAccept"]))}
	state.Bids = new(len(allBids))
	state.Timestamps.WinnerSelected = q.nowISO()
	state.AcceptPending = &v1alpha1.ContractAcceptPending{BidURI: winner.URI}
	r.writeStatus(ctx, v1alpha1.ComputeContractRequesting, "the winner is selected; accepting", state, allBids, chosen, "accepting", 1)

	accept, found, err := q.existingSignedRecord(ctx, market.AcceptNSID, acceptRKey)
	if err != nil {
		return fmt.Errorf("read the market.accept record: %w", err)
	}
	if !found && q.cfg.Spec.SuspendParticipation {
		return ErrParticipationSuspended
	}
	if !found {
		var submitEvent string
		submitEvent, err = r.untilRelease(ctx, q.submitEventRef)
		if err != nil {
			return err
		}
		accept, err = market.CreateAcceptAt(ctx, q.repo, q.repoDID, acceptRKey, market.AcceptOptions{
			RFP:         market.RecordRef{URI: rfpURI, CID: rfpCID},
			Bid:         market.RecordRef{URI: winner.URI, CID: winner.CID},
			SubmitEvent: submitEvent,
			Ingress:     q.ingressRecords(ctx),
			CreatedAt:   q.nowISO(),
		}, q.signer)
		if err != nil {
			return fmt.Errorf("create the market.accept record: %w", err)
		}
	}
	state.AcceptURI = accept.URI
	state.Accept = &commonv1.StrongRef{URI: accept.URI, CID: accept.CID}
	state.AcceptPending = &v1alpha1.ContractAcceptPending{BidURI: winner.URI}
	state.Phase = v1alpha1.ContractAccepted
	state.Timestamps.Accepted = q.nowISO()
	r.writeStatus(ctx, v1alpha1.ComputeContractRequesting, "the accept is submitted; waiting for the receipt", state, allBids, chosen, "awaitingReceipt", 1)
	r.ensureAssociation(ctx, state.Accept, state)

	receipt, err := q.submitAcceptAndWait(ctx, winnerBidder, accept.URI, accept.CID, winner.DID)
	if err != nil {
		return err
	}
	if receipt == nil {
		return fmt.Errorf("no receipt arrived within %s", q.cfg.ReceiptTimeout)
	}
	return r.acceptReceipt(ctx, state, accept.Record, accept.URI, receipt)
}

const (
	submitRetryFirst = 500 * time.Millisecond

	submitRetryMax = 4 * time.Second
)

func submitRFPWithin(ctx context.Context, m MarketClient, bidder Bidder, req SubmitRFPRequest) ([]BidRecord, error) {
	wait := submitRetryFirst
	for {
		bids, err := m.SubmitRFP(ctx, bidder, req)
		if err == nil || !errors.Is(err, ErrBidderUnreachable) {
			return bids, err
		}
		t := time.NewTimer(wait)
		select {
		case <-ctx.Done():
			t.Stop()
			return nil, err
		case <-t.C:
		}
		wait = min(2*wait, submitRetryMax)
	}
}

func (r *contractRunner) solicitBids(ctx context.Context, bidders []Bidder, rfpURI, rfpCID string) {
	for _, bidder := range bidders {
		if bidder.Endpoint == "" {
			continue
		}
		r.q.wg.Go(func() {
			cctx, cancel := context.WithTimeout(ctx, max(r.q.cfg.BidWindow, 5*time.Second))
			defer cancel()
			bids, err := submitRFPWithin(cctx, r.q.market, bidder, SubmitRFPRequest{URI: rfpURI, CID: rfpCID})
			if err != nil {
				r.log().Warn("requester_submit_rfp_failed", "bidder", bidder.DID, "err", err)
				return
			}
			for _, bid := range bids {
				r.q.queueBid(rfpURI, bid)
			}
		})
	}
}

func (r *contractRunner) runGuestAndHold(ctx context.Context, state *v1alpha1.NodeClaimContract, receiptURI, bidderDID string) error {
	if err := r.waitGuest(ctx, state, receiptURI); err != nil {
		return err
	}
	if r.released() {
		return r.releaseContract(ctx, state, bidderDID)
	}
	if state.Termination != nil {
		return nil
	}
	if r.secrets != nil && r.secretsFetched() == "" && !r.q.cfg.SealedSecrets() {
		timer := time.NewTimer(DefaultVMReadyTimeout)
		defer timer.Stop()
		select {
		case <-r.secrets.fetched:
			r.setSecretsFetched(r.secrets.fetchedAtValue())
		case <-timer.C:
			return fmt.Errorf("the guest did not fetch its join secrets within %s", DefaultVMReadyTimeout)
		case <-ctx.Done():
			return ctx.Err()
		}
	}
	state.Phase = v1alpha1.ContractNetwork
	state.Outcome = v1alpha1.ContractOutcomeComplete
	if state.Timestamps.Network == "" {
		state.Timestamps.Network = r.q.nowISO()
	}
	r.writeStatus(ctx, v1alpha1.ComputeContractHeld, "held: the guest is on the network; set spec.release to release it", state, r.currentBids(), r.currentChosen(), "held", 1)
	return r.waitForRelease(ctx, state, receiptURI, bidderDID)
}

func (r *contractRunner) resumeAccepted(ctx context.Context, state *v1alpha1.NodeClaimContract) error {
	bidderDID := winnerDID(state)
	if bidderDID == "" {
		return errors.New("the recorded contract names no winning bidder")
	}
	if err := r.untilSettled(ctx, "resumeAccepted", func() error { return r.q.recordedWinnerAllowed(ctx, state, r.contractPolicyName()) }); err != nil {
		if !errors.Is(err, ErrPolicyRejected) {
			return fmt.Errorf("the recorded accept is not resubmitted: %w", err)
		}
		state.Outcome = v1alpha1.ContractOutcomePolicyRejected
		state.Error = "the recorded accept is not resubmitted: " + err.Error()
		if releaseErr := r.releaseDeniedAccept(ctx, state, bidderDID); releaseErr != nil {
			return fmt.Errorf("%s; releasing it: %w", state.Error, releaseErr)
		}
		return errors.New(state.Error)
	}
	bidder, ok := r.q.bidderForDID(ctx, bidderDID)
	if !ok {
		return fmt.Errorf("no endpoint is known for the winning bidder %s", bidderDID)
	}
	acceptRef := market.RecordRef{URI: state.Accept.URI, CID: state.Accept.CID}
	accept, err := r.q.repo.Resolve(ctx, acceptRef)
	if err != nil {
		return fmt.Errorf("read the recorded market.accept record: %w", err)
	}
	r.ensureAssociation(ctx, state.Accept, state)
	r.writeStatus(ctx, v1alpha1.ComputeContractRequesting, "the accept is recorded; waiting for the receipt", state, r.currentBids(), r.currentChosen(), "awaitingReceipt", 1)
	receipt, err := r.q.submitAcceptAndWait(ctx, bidder, acceptRef.URI, acceptRef.CID, bidderDID)
	if err != nil {
		return err
	}
	if receipt == nil {
		return fmt.Errorf("no receipt arrived within %s", r.q.cfg.ReceiptTimeout)
	}
	return r.acceptReceipt(ctx, state, accept.Value, acceptRef.URI, receipt)
}

const DeniedAcceptReceiptAttempts = 4

func acceptedWithoutReceipt(state *v1alpha1.NodeClaimContract) bool {
	return state != nil && state.Receipt == nil && state.Accept != nil && state.Accept.URI != "" && winnerDID(state) != ""
}

func (r *contractRunner) releaseDeniedAccept(ctx context.Context, state *v1alpha1.NodeClaimContract, bidderDID string) error {
	r.captureReleaseRequest()
	reason := cmp.Or(state.Error, "the contract is released")
	for attempt := 1; !r.recoverReceipt(ctx, state); attempt++ {
		if attempt > DeniedAcceptReceiptAttempts {
			r.captureReleaseRequest()
			release := r.releaseStatus()
			if release == nil {
				release = &v1alpha1.ComputeContractRelease{}
			}
			release.LastError = fmt.Sprintf("no receipt arrived for the accept %s after %d waits of %s; the guest the bidder may have provisioned stays until a release finds the receipt: run `socialweb-computer run request release %s` (spec.release) again, or the bidder ends it at its own TTL",
				state.Accept.URI, DeniedAcceptReceiptAttempts, r.q.cfg.ReceiptTimeout, r.name)
			r.setRelease(release)
			r.writeStatus(ctx, v1alpha1.ComputeContractReleasing, reason+"; "+release.LastError, state, r.currentBids(), r.currentChosen(), "releaseFailed", int32(attempt-1))
			return errors.New(release.LastError)
		}
		release := r.releaseStatus()
		if release == nil {
			release = &v1alpha1.ComputeContractRelease{}
		}
		release.Attempts = attempt
		release.LastError = fmt.Sprintf("waiting for the bidder's receipt of the accept to release its guest (attempt %d of %d)", attempt, DeniedAcceptReceiptAttempts)
		r.setRelease(release)
		r.writeStatus(ctx, v1alpha1.ComputeContractReleasing, reason+"; "+release.LastError, state, r.currentBids(), r.currentChosen(), "releasing", int32(attempt))
		receipt, err := r.q.waitReceiptWithin(ctx, bidderDID, state.Accept.URI, state.Accept.CID, r.q.cfg.ReceiptTimeout)
		if err != nil {
			return err
		}
		if receipt != nil && r.recoverReceipt(ctx, state) {
			break
		}
	}
	return r.releaseContract(ctx, state, bidderDID)
}

func (q *Requester) recordedWinnerAllowed(ctx context.Context, state *v1alpha1.NodeClaimContract, policyName string) error {
	bidderDID := winnerDID(state)
	if bidderDID == "" || state.Policy == nil || state.Policy.URI == "" || state.RFP == nil || state.RFP.URI == "" ||
		state.Bid == nil || state.Bid.URI == "" {
		return errors.New("the recorded contract names no winner, policy, RFP or bid")
	}
	gate, err := q.attachedPolicy(ctx, market.StrongRef{Type: market.StrongRefNSID, URI: state.Policy.URI, CID: state.Policy.CID},
		PolicyWorkflowName(policyName))
	if err != nil {
		return err
	}
	decision := q.policy.Evaluate(ctx, gate.Input(PolicyInput{
		SelfDid:          q.marketDID,
		SubjectDid:       bidderDID,
		CounterpartyDid:  bidderDID,
		RootRequesterDid: q.marketDID,
		RFP:              &PolicyRef{URI: state.RFP.URI, CID: state.RFP.CID},
		Bid:              &PolicyRef{URI: state.Bid.URI, CID: state.Bid.CID},
	}))
	switch {
	case decision.Allow:
		return nil
	case !decision.Completed:
		return retryable(errors.New("the policy run for the recorded winner " + bidderDID + " did not complete: " + decision.Reason))
	}
	return fmt.Errorf("%w: the recorded winner %s: %s", ErrPolicyRejected, bidderDID, decision.Reason)
}

type retryableError struct{ error }

func (e retryableError) Unwrap() error { return e.error }

func retryable(err error) error {
	if err == nil {
		return nil
	}
	return retryableError{err}
}

func isRetryable(err error) bool {
	var target retryableError
	return errors.As(err, &target)
}

func (r *contractRunner) untilSettled(ctx context.Context, site string, try func() error) error {
	err := try()
	if err == nil || !isRetryable(err) || ctx.Err() != nil {
		return err
	}
	wake, cancel := r.q.watchBinds()
	defer cancel()
	giveUp := time.Now().Add(TransientResumeGiveUp)
	r.log().Info("requester_contract_waits_transient", "site", site, "err", err.Error(),
		"reason", "a transient start-up failure keeps the contract in its phase; a Vault, MarketIdentity or AtprotoRepo bind or the backoff wakes the next try; no success within "+TransientResumeGiveUp.String()+" fails it")
	for failures := 1; ; failures++ {
		wait := time.NewTimer(netretry.Delay(failures))
		select {
		case <-ctx.Done():
			wait.Stop()
			return err
		case <-r.releaseCh:
			wait.Stop()
			return err
		case <-wake:
			wait.Stop()
		case <-wait.C:
		}
		err = try()
		switch {
		case err == nil || !isRetryable(err):
			r.log().Info("requester_contract_transient_settled", "site", site, "attempts", failures+1, "err", fmt.Sprint(err))
			return err
		case ctx.Err() != nil:
			return err
		case !time.Now().Before(giveUp):
			r.log().Warn("requester_contract_transient_gave_up", "site", site, "attempts", failures+1, "err", err.Error())
			return err
		}
	}
}

func (r *contractRunner) contractPolicyName() string {
	return cmp.Or(strings.TrimSpace(r.snapshot().Spec.Policy), r.q.cfg.PolicyWorkflow, DefaultPolicyWorkflow)
}

func (r *contractRunner) recoverReceipt(ctx context.Context, state *v1alpha1.NodeClaimContract) bool {
	bidderDID := winnerDID(state)
	if state == nil || state.Receipt != nil || state.Accept == nil || state.Accept.URI == "" || bidderDID == "" {
		return false
	}
	receipt, err := r.q.readReceipt(ctx, bidderDID, state.Accept.URI, state.Accept.CID)
	if err != nil || receipt == nil {
		r.log().Info("requester_release_no_receipt", "accept", state.Accept.URI, "bidder", bidderDID, "err", fmt.Sprint(err))
		return false
	}
	accept, err := r.q.repo.Resolve(ctx, market.RecordRef{URI: state.Accept.URI, CID: state.Accept.CID})
	if err != nil {
		r.log().Warn("requester_release_accept_unread", "accept", state.Accept.URI, "err", err)
		return false
	}
	if verification := market.VerifyReceipt(ctx, receipt.Value, receipt.URI, accept.Value, state.Accept.URI, r.q.receiptOptions()); !verification.OK {
		r.log().Warn("requester_release_receipt_invalid", "receipt", receipt.URI)
		return false
	}
	state.Receipt = &commonv1.StrongRef{URI: receipt.URI, CID: receipt.CID}
	state.SubmitEventRef = stringOf(receipt.Value["submitEvent"])
	state.ReceiptOk = new(true)
	r.setState(state)
	r.log().Info("requester_release_found_receipt", "receipt", receipt.URI, "bidder", bidderDID)
	return true
}

func (r *contractRunner) acceptReceipt(ctx context.Context, state *v1alpha1.NodeClaimContract, acceptRecord map[string]any, acceptURI string, receipt *market.ResolvedRecord) error {
	verification := market.VerifyReceipt(ctx, receipt.Value, receipt.URI, acceptRecord, acceptURI, r.q.receiptOptions())
	state.Receipt = &commonv1.StrongRef{URI: receipt.URI, CID: receipt.CID}
	state.SubmitEventRef = stringOf(receipt.Value["submitEvent"])
	state.ReceiptOk = new(verification.OK)
	state.Phase = v1alpha1.ContractReceipt
	state.Timestamps.Receipt = r.q.nowISO()
	if !verification.OK {
		state.Outcome = v1alpha1.ContractOutcomeReceiptInvalid
		state.Error = "the receipt signature or accept binding is invalid"
		r.writeStatus(ctx, v1alpha1.ComputeContractFailed, state.Error, state, r.currentBids(), r.currentChosen(), "receiptInvalid", 1)
		return errors.New(state.Error)
	}
	r.writeStatus(ctx, v1alpha1.ComputeContractRequesting, "the receipt is verified; waiting for the guest", state, r.currentBids(), r.currentChosen(), "awaitingGuestOnNetwork", 1)
	if state.Termination != nil {
		return nil
	}
	r.ensureAtircToken(ctx, r.snapshot(), state)
	return r.runGuestAndHold(ctx, state, receipt.URI, winnerDID(state))
}

func (r *contractRunner) resumeReceipt(ctx context.Context, state *v1alpha1.NodeClaimContract) error {
	if r.refreshTermination(ctx, state) {
		r.writeStatus(ctx, v1alpha1.ComputeContractReleased, terminationMessage(state.Termination), state, r.currentBids(), r.currentChosen(), "terminated", 1)
		return nil
	}
	bidderDID := winnerDID(state)
	if r.released() {
		return r.releaseContract(ctx, state, bidderDID)
	}
	if err := r.untilSettled(ctx, "resumeReceipt", func() error { return r.q.recordedWinnerAllowed(ctx, state, r.contractPolicyName()) }); err != nil {
		if r.released() && !errors.Is(err, ErrPolicyRejected) {
			return r.releaseContract(ctx, state, bidderDID)
		}
		if !errors.Is(err, ErrPolicyRejected) {
			return fmt.Errorf("the held contract is not resumed: %w", err)
		}
		state.Outcome = v1alpha1.ContractOutcomePolicyRejected
		state.Error = "the held contract is not resumed: " + err.Error()
		if releaseErr := r.releaseContract(ctx, state, bidderDID); releaseErr != nil {
			return fmt.Errorf("%s; releasing it: %w", state.Error, releaseErr)
		}
		return errors.New(state.Error)
	}
	if r.released() {
		return r.releaseContract(ctx, state, bidderDID)
	}
	r.ensureAssociation(ctx, state.Accept, state)
	receiptRef := market.RecordRef{URI: state.Receipt.URI, CID: state.Receipt.CID}
	verified, err := r.verifyRecordedReceipt(ctx, state, receiptRef)
	if err != nil {
		if state.ReceiptOk == nil || !*state.ReceiptOk {
			return err
		}
	} else if !verified {
		state.ReceiptOk = new(false)
		state.Outcome = v1alpha1.ContractOutcomeReceiptInvalid
		state.Error = "the receipt signature or accept binding is invalid"
		r.writeStatus(ctx, v1alpha1.ComputeContractFailed, state.Error, state, r.currentBids(), r.currentChosen(), "receiptInvalid", 1)
		return errors.New(state.Error)
	} else {
		state.ReceiptOk = new(true)
	}
	if r.released() {
		return r.releaseContract(ctx, state, bidderDID)
	}
	if state.Phase != v1alpha1.ContractNetwork && state.Phase != v1alpha1.ContractReleased {
		state.Phase = v1alpha1.ContractReceipt
	}
	if state.Timestamps == nil {
		state.Timestamps = &v1alpha1.ContractTimestamps{}
	}
	if state.Timestamps.Receipt == "" {
		state.Timestamps.Receipt = r.q.nowISO()
	}
	r.ensureAtircToken(ctx, r.snapshot(), state)
	return r.runGuestAndHold(ctx, state, receiptRef.URI, bidderDID)
}

func (r *contractRunner) resumeReleased(ctx context.Context, state *v1alpha1.NodeClaimContract) {
	terminated := r.refreshReleasedTermination(ctx, state)
	message := "released"
	step := "released"
	if terminated {
		message = terminationMessage(state.Termination)
		step = "terminated"
		r.log().Info("requester_contract_termination_resumed", "receipt", state.Receipt.URI, "event", state.Termination.Event.URI, "reason", state.Termination.Reason)
	}
	r.writeStatus(ctx, v1alpha1.ComputeContractReleased, message, state, r.currentBids(), r.currentChosen(), step, 1)
}

func (r *contractRunner) refreshReleasedTermination(ctx context.Context, state *v1alpha1.NodeClaimContract) bool {
	terminated, answered := r.checkTermination(ctx, state)
	if !terminated && answered {
		r.terminationNotReported.Store(true)
	}
	return terminated
}

func (r *contractRunner) refreshTermination(ctx context.Context, state *v1alpha1.NodeClaimContract) bool {
	terminated, _ := r.checkTermination(ctx, state)
	return terminated
}

func (r *contractRunner) checkTermination(ctx context.Context, state *v1alpha1.NodeClaimContract) (bool, bool) {
	if state == nil {
		return false, false
	}
	if state.Termination != nil {
		state.Phase = v1alpha1.ContractReleased
		return true, true
	}
	if state.Receipt == nil || state.Receipt.URI == "" || r.q.market == nil {
		return false, false
	}
	bidderDID := winnerDID(state)
	if bidderDID == "" {
		return false, false
	}
	readCtx, cancel := context.WithTimeout(ctx, DefaultHTTPTimeout)
	defer cancel()
	readStart := time.Now()
	event, err := r.q.market.FindTermination(readCtx, bidderDID, state.Receipt.URI)
	readMs := time.Since(readStart).Milliseconds()
	r.log().Debug("requester_termination_checked", "bidder", bidderDID, "receipt", state.Receipt.URI, "found", event != nil, "readMs", readMs, "err", fmt.Sprint(err))
	r.q.noteFirstResume(r.name, readMs)
	if err != nil {
		if ctx.Err() == nil {
			r.log().Warn("requester_termination_resume_read_failed", "bidder", bidderDID, "receipt", state.Receipt.URI, "err", err)
		}
		return false, false
	}
	if event == nil {
		return false, true
	}
	parsed, err := r.q.parseGuestEvent(readCtx, *event, state.Receipt.URI)
	if err != nil {
		r.log().Warn("requester_termination_resume_refused", "uri", event.Event.URI, "err", err)
		return false, false
	}
	if parsed.Kind != market.ComputeEventsVMTerminatedNSID {
		return false, true
	}
	applyTermination(state, parsed.Termination, r.q.nowISO())
	r.log().Info("requester_termination_resumed", "bidder", bidderDID, "receipt", state.Receipt.URI, "event", parsed.Event.URI, "reason", parsed.Termination.Reason)
	return true, true
}

func (r *contractRunner) verifyRecordedReceipt(ctx context.Context, state *v1alpha1.NodeClaimContract, receiptRef market.RecordRef) (bool, error) {
	receipt, err := r.q.repo.Resolve(ctx, receiptRef)
	if err != nil {
		return false, fmt.Errorf("read the recorded market.receipt record: %w", err)
	}
	if state.SubmitEventRef == "" {
		state.SubmitEventRef = stringOf(receipt.Value["submitEvent"])
	}
	if state.Accept == nil || state.Accept.URI == "" {
		return true, nil
	}
	accept, err := r.q.repo.Resolve(ctx, market.RecordRef{URI: state.Accept.URI, CID: state.Accept.CID})
	if err != nil {
		return false, fmt.Errorf("read the recorded market.accept record: %w", err)
	}
	verification := market.VerifyReceipt(ctx, receipt.Value, receipt.URI, accept.Value, state.Accept.URI, r.q.receiptOptions())
	return verification.OK, nil
}

func (r *contractRunner) collectBids(ctx context.Context, rfpURI string, workflow PolicyGate, only []string) (*BidRecord, []v1alpha1.ContractBid, error) {
	q := r.q
	window := time.NewTimer(max(q.cfg.BidWindow, 0))
	defer window.Stop()
	byURI := map[string]BidRecord{}
	deny := r.snapshot().Spec.DenyBidderDids
	add := func(bids []BidRecord) {
		for _, bid := range bids {
			if bid.URI == "" || slices.Contains(deny, bid.DID) {
				continue
			}
			if len(only) > 0 && !slices.Contains(only, bid.DID) {
				r.log().Info("requester_bid_outside_contract_bidders", "bidder", bid.DID, "bid", bid.URI)
				continue
			}
			byURI[bid.URI] = bid
		}
	}
	var selected *BidRecord
	open := q.cfg.BidWindow > 0
	for selected == nil {
		arrived := q.bids.arrived()
		add(q.bids.take(rfpURI))
		if candidate := r.freeWinner(ctx, byURI, workflow); candidate != nil {
			selected = candidate
			break
		}
		if !open {
			break
		}
		select {
		case <-ctx.Done():
			return nil, nil, ctx.Err()
		case <-r.releaseCh:
			return nil, nil, nil
		case <-window.C:
			open = false
		case <-arrived:
		}
	}
	add(q.bids.take(rfpURI))
	if selected == nil {
		selected = r.bestBid(ctx, byURI, workflow)
	}
	records := make([]v1alpha1.ContractBid, 0, len(byURI))
	for _, bid := range byURI {
		entry := v1alpha1.ContractBid{
			Did:       bid.DID,
			URI:       bid.URI,
			Verdict:   v1alpha1.ContractBidConsidered,
			ArrivedAt: bid.arrivedAtISO(),
		}
		if !bid.Free {
			entry.Cost = int64(bid.Cost)
		}
		switch {
		case selected != nil && selected.URI == bid.URI:
			entry.Verdict = v1alpha1.ContractBidChosen
		case !q.bidAllowedWithPolicy(ctx, bid, workflow):
			entry.Verdict = v1alpha1.ContractBidPolicyRejected
		case selected != nil:
			entry.Verdict = v1alpha1.ContractBidNotChosen
		default:
			entry.Verdict = v1alpha1.ContractBidDenied
		}
		records = append(records, entry)
		if len(records) >= MaxRecordBids {
			break
		}
	}
	return selected, records, nil
}

func (r *contractRunner) freeWinner(ctx context.Context, bids map[string]BidRecord, workflow PolicyGate) *BidRecord {
	for _, bid := range bids {
		bidCopy := bid
		if err := r.q.classifyBid(ctx, &bidCopy); err != nil {
			continue
		}
		if bidCopy.Free && r.q.bidAllowedWithPolicy(ctx, bidCopy, workflow) {
			return &bidCopy
		}
	}
	return nil
}

func (r *contractRunner) bestBid(ctx context.Context, bids map[string]BidRecord, workflow PolicyGate) *BidRecord {
	var best *BidRecord
	for _, bid := range bids {
		bidCopy := bid
		if err := r.q.classifyBid(ctx, &bidCopy); err != nil {
			continue
		}
		if !r.q.bidAllowedWithPolicy(ctx, bidCopy, workflow) {
			continue
		}
		if best == nil || bidCopy.Cost < best.Cost {
			best = &bidCopy
		}
	}
	return best
}

func (r *contractRunner) waitGuest(ctx context.Context, state *v1alpha1.NodeClaimContract, receiptURI string) error {
	events, pending, closeEvents := r.q.registerEventSink(receiptURI)
	defer closeEvents()
	addressDeadline := time.NewTimer(DefaultVMReadyTimeout)
	defer addressDeadline.Stop()
	hostKeysDeadline := time.NewTimer(DefaultHostKeysTimeout)
	defer hostKeysDeadline.Stop()
	addressLate, hostKeysLate := false, false
	seen := map[string]bool{}
	seedGuestEventSeen(seen, state)
	handle := func(event MarketEvent) bool {
		if guestEventSeen(seen, event) || isContractEventNSID(market.NSIDFromURI(event.Payload.URI)) {
			return false
		}
		parsed, err := r.q.parseGuestEvent(ctx, event, receiptURI)
		if err != nil {
			r.log().Warn("requester_guest_event_refused", "uri", event.Event.URI, "err", err)
			return false
		}
		switch parsed.Kind {
		case market.ComputeEventsVMOnNetworkNSID:
			if parsed.Tag != "" {
				if !recordNetwork(state, parsed) {
					return false
				}
				r.log().Info("requester_guest_network", "tag", parsed.Tag, "address", parsed.Address)
				break
			}
			if state.VMAddress != "" {
				return false
			}
			state.VMAddress = parsed.Address
			state.Phase = v1alpha1.ContractNetwork
			if state.Timestamps == nil {
				state.Timestamps = &v1alpha1.ContractTimestamps{}
			}
			state.Timestamps.Network = r.q.nowISO()
			r.log().Info("requester_guest_on_network", "address", parsed.Address)
		case market.ComputeEventsVMHostKeysNSID:
			if state.HostKeys != nil && guestEventRecorded(state.HostKeys.Event, state.HostKeys.Payload, parsed) {
				return false
			}
			state.HostKeys = &v1alpha1.ContractHostKeys{
				Keys:       contractHostKeys(parsed.HostKeys),
				ObservedAt: parsed.ObservedAt,
				Event:      commonv1.StrongRef{URI: parsed.Event.URI, CID: parsed.Event.CID},
				Payload:    commonv1.StrongRef{URI: parsed.Payload.URI, CID: parsed.Payload.CID},
			}
			r.log().Info("requester_guest_host_keys", "keys", len(parsed.HostKeys))
		case market.ComputeEventsVMTerminatedNSID:
			applyTermination(state, parsed.Termination, r.q.nowISO())
			r.writeStatus(ctx, v1alpha1.ComputeContractReleased, terminationMessage(parsed.Termination), state, r.currentBids(), r.currentChosen(), "terminated", 1)
			return true
		case market.ComputeEventsVMSandboxNSID:
			if parsed.Sandbox == nil {
				return false
			}
			state.Sandbox = parsed.Sandbox
			r.log().Info("requester_guest_sandbox", "exit", parsed.Sandbox.ExitCode, "timedOut", parsed.Sandbox.TimedOut,
				"stdout", parsed.Sandbox.Stdout, "stderr", parsed.Sandbox.Stderr)
		case market.ComputeEventsGuestSecretsRequestedNSID, market.ComputeEventsGuestSecretsAppliedNSID:
			if !applySecretsEvent(state, parsed) {
				return false
			}
			r.log().Info("requester_guest_secrets", "event", payloadKind(parsed.Kind), "via", parsed.SecretsVia)
		case market.ComputeEventsGuestRequesterDIDNSID:
			r.onGuestRequester(ctx, state, parsed)
			return false
		}
		r.writeStatus(ctx, v1alpha1.ComputeContractRequesting, "the guest reported "+payloadKind(parsed.Kind), state, r.currentBids(), r.currentChosen(), "awaitingGuestOnNetwork", 1)
		if parsed.Kind == market.ComputeEventsGuestSecretsRequestedNSID {
			r.sealSecrets(ctx, state)
		}
		return false
	}
	if r.released() || slices.ContainsFunc(pending, handle) {
		return nil
	}
	var retry eventRetry
	defer retry.stop()
	retryC := retry.after(r.q.drainEvents(ctx, r.name, receiptURI))
	for {
		if state.Termination != nil {
			return nil
		}
		if state.VMAddress != "" && state.HostKeys != nil {
			return nil
		}
		if sandboxAddress(state.VMAddress) && state.Sandbox != nil {
			return nil
		}
		if r.released() {
			return nil
		}
		if state.VMAddress == "" && addressLate {
			state.Outcome = v1alpha1.ContractOutcomeNetworkTimeout
			return fmt.Errorf("the guest did not come on network within %s", DefaultVMReadyTimeout)
		}
		if state.VMAddress != "" && state.HostKeys == nil && !sandboxAddress(state.VMAddress) && hostKeysLate {
			return fmt.Errorf("the guest reported no ssh host keys within %s", DefaultHostKeysTimeout)
		}
		if sandboxAddress(state.VMAddress) && state.Sandbox == nil && addressLate {
			return fmt.Errorf("the sandbox reported no result within %s", DefaultVMReadyTimeout)
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-r.releaseCh:
			return nil
		case <-addressDeadline.C:
			addressLate = true
		case <-hostKeysDeadline.C:
			hostKeysLate = true
		case event := <-events:
			if handle(event) {
				return nil
			}
		case <-retryC:
			retryC = retry.after(r.q.drainEvents(ctx, r.name, receiptURI))
		case <-r.associationKick:
			r.associationStep(ctx, state)
		}
	}
}

func (r *contractRunner) waitForRelease(ctx context.Context, state *v1alpha1.NodeClaimContract, receiptURI, bidderDID string) error {
	events, pending, closeEvents := r.q.registerEventSink(receiptURI)
	defer closeEvents()
	seen := map[string]bool{}
	seedGuestEventSeen(seen, state)
	handle := func(event MarketEvent) bool {
		if guestEventSeen(seen, event) {
			return false
		}
		nsid := market.NSIDFromURI(event.Payload.URI)
		switch nsid {
		case market.ComputeEventsVMTerminatedNSID, market.ComputeEventsVMOnNetworkNSID,
			market.ComputeEventsGuestSecretsRequestedNSID, market.ComputeEventsGuestSecretsAppliedNSID,
			market.ComputeEventsGuestRequesterDIDNSID:
		default:
			return false
		}
		parsed, err := r.q.parseGuestEvent(ctx, event, receiptURI)
		if err != nil {
			if nsid == market.ComputeEventsVMOnNetworkNSID {
				r.log().Warn("requester_guest_event_refused", "uri", event.Event.URI, "err", err)
			}
			return false
		}
		if parsed.Kind == market.ComputeEventsVMOnNetworkNSID {
			if parsed.Tag != "" && recordNetwork(state, parsed) {
				r.log().Info("requester_guest_network", "tag", parsed.Tag, "address", parsed.Address)
				r.writeStatus(ctx, v1alpha1.ComputeContractHeld, "held: the guest is on the network; set spec.release to release it", state, r.currentBids(), r.currentChosen(), "held", 1)
			}
			return false
		}
		if parsed.Kind == market.ComputeEventsGuestRequesterDIDNSID {
			r.onGuestRequester(ctx, state, parsed)
			return false
		}
		if parsed.Kind == market.ComputeEventsGuestSecretsRequestedNSID || parsed.Kind == market.ComputeEventsGuestSecretsAppliedNSID {
			if applySecretsEvent(state, parsed) {
				r.log().Info("requester_guest_secrets", "event", payloadKind(parsed.Kind), "via", parsed.SecretsVia)
				r.writeStatus(ctx, v1alpha1.ComputeContractHeld, "held: the guest is on the network; set spec.release to release it", state, r.currentBids(), r.currentChosen(), "held", 1)
				if parsed.Kind == market.ComputeEventsGuestSecretsRequestedNSID {
					r.sealSecrets(ctx, state)
				}
			}
			return false
		}
		if parsed.Kind != market.ComputeEventsVMTerminatedNSID {
			return false
		}
		applyTermination(state, parsed.Termination, r.q.nowISO())
		r.writeStatus(ctx, v1alpha1.ComputeContractReleased, terminationMessage(parsed.Termination), state, r.currentBids(), r.currentChosen(), "terminated", 1)
		return true
	}
	if slices.ContainsFunc(pending, handle) {
		return nil
	}
	var retry eventRetry
	defer retry.stop()
	retryC := retry.after(r.q.drainEvents(ctx, r.name, receiptURI))
	for {
		if state.Termination != nil {
			return nil
		}
		select {
		case <-ctx.Done():
			if r.q.stopping() {
				r.log().Info("requester_contract_left_for_restart")
				return nil
			}
			return ctx.Err()
		case <-r.releaseCh:
			return r.releaseContract(ctx, state, bidderDID)
		case event := <-events:
			if handle(event) {
				return nil
			}
		case <-retryC:
			retryC = retry.after(r.q.drainEvents(ctx, r.name, receiptURI))
		case <-r.associationKick:
			r.associationStep(ctx, state)
		}
	}
}

func receiptURIOf(state *v1alpha1.NodeClaimContract) string {
	if state == nil {
		return ""
	}
	if state.Receipt != nil && state.Receipt.URI != "" {
		return state.Receipt.URI
	}
	return state.ReceiptURI
}

func (r *contractRunner) finishRelease(ctx context.Context, state *v1alpha1.NodeClaimContract, release *v1alpha1.ComputeContractRelease, message string) {
	state.Release.OK = true
	state.Phase = v1alpha1.ContractReleased
	if state.Timestamps != nil && state.Timestamps.Released == "" {
		state.Timestamps.Released = r.q.nowISO()
	}
	release.LastError = ""
	r.setRelease(release)
	r.log().Info("requester_released", "receipt", receiptURIOf(state), "attempts", release.Attempts, "message", message)
	r.writeStatus(ctx, v1alpha1.ComputeContractReleased, message, state, r.currentBids(), r.currentChosen(), "released", int32(release.Attempts))
	r.deleteContractRecords(ctx)
}

func (r *contractRunner) captureReleaseRequest() {
	requested := r.snapshot().Spec.ReleaseRequestedAt
	r.mu.Lock()
	r.releaseRequested = requested
	r.mu.Unlock()
}

func (r *contractRunner) releaseRequestedAt() string {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.releaseRequested
}

func guestMayRun(state *v1alpha1.NodeClaimContract) bool {
	return state != nil && state.Receipt != nil && state.ReceiptOk != nil && *state.ReceiptOk && state.Termination == nil &&
		(state.Release == nil || !state.Release.OK)
}

func (r *contractRunner) applyReleaseBacklog(ctx context.Context, state *v1alpha1.NodeClaimContract) {
	receiptURI := receiptURIOf(state)
	if receiptURI == "" {
		return
	}
	events, pending, closeEvents := r.q.registerEventSink(receiptURI)
	var skipped []MarketEvent
	defer func() {
		closeEvents()
		for _, event := range skipped {
			r.q.dispatchEvent(receiptURI, event)
		}
	}()
	apply := func(event MarketEvent) {
		switch market.NSIDFromURI(event.Payload.URI) {
		case market.ComputeEventsGuestRequesterDIDNSID, market.ComputeEventsVMTerminatedNSID:
		default:
			skipped = append(skipped, event)
			return
		}
		parsed, err := r.q.parseGuestEvent(ctx, event, receiptURI)
		if err != nil {
			r.log().Warn("requester_guest_event_refused", "uri", event.Event.URI, "err", err)
			return
		}
		switch parsed.Kind {
		case market.ComputeEventsGuestRequesterDIDNSID:
			r.recordGuestRequester(ctx, state, parsed)
		case market.ComputeEventsVMTerminatedNSID:
			if applyTermination(state, parsed.Termination, r.q.nowISO()) {
				r.log().Info("requester_release_found_termination", "event", parsed.Event.URI, "reason", parsed.Termination.Reason)
			}
		}
	}
	for _, event := range pending {
		apply(event)
	}
	_ = r.q.drainEvents(ctx, r.name, receiptURI)
	for {
		select {
		case event := <-events:
			apply(event)
		default:
			return
		}
	}
}

func (r *contractRunner) releaseContract(ctx context.Context, state *v1alpha1.NodeClaimContract, bidderDID string) error {
	q := r.q
	r.captureReleaseRequest()
	if state == nil || state.Receipt == nil {
		r.writeStatus(ctx, v1alpha1.ComputeContractReleased, "released before a VM was held", state, r.currentBids(), r.currentChosen(), "released", 1)
		r.deleteContractRecords(ctx)
		return nil
	}
	if state.Release == nil || !state.Release.OK {
		r.applyReleaseBacklog(ctx, state)
	}
	if state.Termination != nil {
		state.Phase = v1alpha1.ContractReleased
		if state.Timestamps != nil && state.Timestamps.Released == "" {
			state.Timestamps.Released = q.nowISO()
		}
		r.writeStatus(ctx, v1alpha1.ComputeContractReleased, terminationMessage(state.Termination), state, r.currentBids(), r.currentChosen(), "terminated", 1)
		r.deleteContractRecords(ctx)
		return nil
	}
	if state.Release != nil && state.Release.OK {
		state.Phase = v1alpha1.ContractReleased
		if release := r.releaseStatus(); release != nil {
			release.LastError = ""
			r.setRelease(release)
		}
		r.writeStatus(ctx, v1alpha1.ComputeContractReleased, "released", state, r.currentBids(), r.currentChosen(), "released", 1)
		r.deleteContractRecords(ctx)
		return nil
	}

	release := r.releaseStatus()
	if release == nil {
		release = &v1alpha1.ComputeContractRelease{}
	}
	if release.StartedAt == "" {
		release.StartedAt = q.nowISO()
	}
	deadline := q.releaseDeadline(r.snapshot())
	if release.Deadline != "" {
		if parsed, err := time.Parse(time.RFC3339, release.Deadline); err == nil {
			deadline = parsed
		}
	} else {
		release.Deadline = deadline.UTC().Format(time.RFC3339)
	}
	r.setRelease(release)

	bidder, ok := q.winnerBidder(ctx, state, bidderDID)
	if !ok {
		bidder, ok = q.bidderForDID(ctx, bidderDID)
	}
	if !ok {
		bidder = Bidder{}
	}

	r.log().Info("requester_release", "receipt", receiptURIOf(state), "bidder", bidderDID, "deadline", release.Deadline, "attempts", release.Attempts)
	r.writeStatus(ctx, v1alpha1.ComputeContractReleasing, "destroying the guest", state, r.currentBids(), r.currentChosen(), "releasing", int32(release.Attempts))

	r.releaseGuestRequester(ctx, state)
	r.revokeBeforeDestroy(ctx)
	eventRef, eventRecord, err := r.releaseEnvelope(ctx, state)
	for err != nil {
		release.Attempts++
		release.LastError = err.Error()
		r.setRelease(release)
		r.log().Info("requester_release_retry", "attempt", release.Attempts, "stage", "record", "err", release.LastError)
		message := fmt.Sprintf("the destroy event could not be recorded on attempt %d: %s", release.Attempts, release.LastError)
		if q.now().After(deadline) {
			r.writeStatus(ctx, v1alpha1.ComputeContractReleasing, message+"; the deadline passed, the release intent stays recorded", state, r.currentBids(), r.currentChosen(), "releaseFailed", int32(release.Attempts))
			return fmt.Errorf("the destroy event was not recorded by %s: %w", deadline.UTC().Format(time.RFC3339), err)
		}
		r.writeStatus(ctx, v1alpha1.ComputeContractReleasing, message+"; retrying", state, r.currentBids(), r.currentChosen(), "releaseFailed", int32(release.Attempts))
		if err := sleepCtx(ctx, releaseRetryDelay(release.Attempts)); err != nil {
			return err
		}
		if r.settleTerminated(ctx, state) {
			return nil
		}
		eventRef, eventRecord, err = r.releaseEnvelope(ctx, state)
	}
	r.writeStatus(ctx, v1alpha1.ComputeContractReleasing, "the destroy is recorded; sending it", state, r.currentBids(), r.currentChosen(), "releasing", int32(release.Attempts))

	if bidder.Endpoint == "" {
		release.Attempts++
		r.finishRelease(ctx, state, release, "released; the firehose carries the destroy")
		return nil
	}

	for {
		release.Attempts++
		r.setRelease(release)
		r.writeStatus(ctx, v1alpha1.ComputeContractReleasing, fmt.Sprintf("destroying the guest: attempt %d", release.Attempts), state, r.currentBids(), r.currentChosen(), "releasing", int32(release.Attempts))
		status, err := q.market.SubmitEvent(ctx, bidder, SubmitEventRequest{URI: eventRef.URI, CID: eventRef.CID, Record: eventRecord})
		if err == nil && status >= 200 && status <= 299 {
			state.Release.Status = &status
			r.finishRelease(ctx, state, release, "released")
			return nil
		}
		detail := "no answer"
		if status != 0 {
			detail = fmt.Sprintf("HTTP %d", status)
			state.Release.Status = &status
		}
		if err != nil {
			detail = err.Error()
		}
		release.LastError = detail
		r.setRelease(release)
		r.log().Info("requester_release_retry", "attempt", release.Attempts, "err", detail)
		message := fmt.Sprintf("the destroy failed on attempt %d: %s", release.Attempts, detail)
		if q.now().After(deadline) {
			r.writeStatus(ctx, v1alpha1.ComputeContractReleasing, message+"; the deadline passed, the release intent stays recorded", state, r.currentBids(), r.currentChosen(), "releaseFailed", int32(release.Attempts))
			return fmt.Errorf("the guest was not destroyed by %s: %s", deadline.UTC().Format(time.RFC3339), detail)
		}
		r.writeStatus(ctx, v1alpha1.ComputeContractReleasing, message+"; retrying", state, r.currentBids(), r.currentChosen(), "releaseFailed", int32(release.Attempts))
		if err := sleepCtx(ctx, releaseRetryDelay(release.Attempts)); err != nil {
			return err
		}
		if r.settleTerminated(ctx, state) {
			return nil
		}
	}
}

func (r *contractRunner) settleTerminated(ctx context.Context, state *v1alpha1.NodeClaimContract) bool {
	if terminated, _ := r.checkTermination(ctx, state); !terminated || state.Termination == nil {
		return false
	}
	if state.Timestamps != nil && state.Timestamps.Released == "" {
		state.Timestamps.Released = r.q.nowISO()
	}
	if release := r.releaseStatus(); release != nil {
		release.LastError = ""
		r.setRelease(release)
	}
	r.log().Info("requester_release_settled_by_termination", "receipt", receiptURIOf(state), "bidder", winnerDID(state), "reason", state.Termination.Reason)
	r.writeStatus(ctx, v1alpha1.ComputeContractReleased, terminationMessage(state.Termination), state, r.currentBids(), r.currentChosen(), "terminated", 1)
	r.deleteContractRecords(ctx)
	return true
}

func (q *Requester) winnerBidder(ctx context.Context, state *v1alpha1.NodeClaimContract, bidderDID string) (Bidder, bool) {
	if state.Winner == nil || state.Winner.DID != bidderDID {
		return Bidder{}, false
	}
	ref := strings.TrimSpace(state.Winner.SubmitAccept)
	if ref == "" {
		return Bidder{}, false
	}
	endpoint, audience, err := resolveBidderEndpoint(ctx, q.deps.Directory, ref)
	if err != nil || endpoint == "" {
		return Bidder{}, false
	}
	return Bidder{DID: bidderDID, Endpoint: endpoint, Audience: audience}, true
}

func sleepCtx(ctx context.Context, delay time.Duration) error {
	timer := time.NewTimer(delay)
	defer timer.Stop()
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-timer.C:
		return nil
	}
}

func (q *Requester) releaseDeadline(contract v1alpha1.ComputeContract) time.Time {
	base := contract.CreationTimestamp.Time
	if base.IsZero() {
		base = q.now()
	}
	bound := DefaultReleaseBound
	if ttl := contract.Spec.TTLSecondsAfterFinished; ttl != nil && *ttl > 0 {
		bound = time.Duration(*ttl) * time.Second
	}
	return base.Add(bound)
}

func releaseRetryDelay(attempt int) time.Duration {
	delay := DefaultReleaseRetryFirst
	for i := 1; i < attempt; i++ {
		delay *= 2
		if delay >= DefaultReleaseRetryMax {
			return DefaultReleaseRetryMax
		}
	}
	return min(delay, DefaultReleaseRetryMax)
}

func (r *contractRunner) releaseEnvelope(ctx context.Context, state *v1alpha1.NodeClaimContract) (market.RecordRef, map[string]any, error) {
	if state.Release != nil && state.Release.Event.URI != "" {
		ref := market.RecordRef{URI: state.Release.Event.URI, CID: state.Release.Event.CID}
		resolved, err := r.q.repo.Resolve(ctx, ref)
		if err != nil {
			return market.RecordRef{}, nil, fmt.Errorf("read the recorded destroy event: %w", err)
		}
		return ref, resolved.Value, nil
	}
	envelope, payload, err := r.q.destroyEnvelope(ctx, state.Receipt, "released")
	if err != nil {
		return market.RecordRef{}, nil, err
	}
	state.Release = &v1alpha1.ContractRelease{
		Event:   commonv1.StrongRef{URI: envelope.URI, CID: envelope.CID},
		Payload: commonv1.StrongRef{URI: payload.URI, CID: payload.CID},
	}
	return market.RecordRef{URI: envelope.URI, CID: envelope.CID}, envelope.Record, nil
}

func (q *Requester) destroyEnvelope(ctx context.Context, receipt *commonv1.StrongRef, reason string) (market.SignedRecord, market.SignedRecord, error) {
	payloadRKey := market.ReleasePayloadRKey(receipt.URI)
	payload, foundPayload, err := q.existingSignedRecord(ctx, market.ComputeEventsVMDeleteNSID, payloadRKey)
	if err != nil {
		return market.SignedRecord{}, market.SignedRecord{}, fmt.Errorf("read the destroy payload: %w", err)
	}
	if !foundPayload {
		payloadRecord, err := market.EncodeRecord(market.ComputeEventsVMDelete{Reason: reason, CreatedAt: q.nowISO()})
		if err != nil {
			return market.SignedRecord{}, market.SignedRecord{}, fmt.Errorf("encode the destroy payload: %w", err)
		}
		payload, err = market.CreateSignedRecordAt(ctx, q.repo, q.repoDID, market.ComputeEventsVMDeleteNSID, payloadRKey, payloadRecord, q.signer)
		if err != nil {
			return market.SignedRecord{}, market.SignedRecord{}, fmt.Errorf("create the destroy payload: %w", err)
		}
	}
	eventRKey := sessionRKey("event", receipt.URI)
	envelope, foundEvent, err := q.existingSignedRecord(ctx, market.EventNSID, eventRKey)
	if err != nil {
		return market.SignedRecord{}, market.SignedRecord{}, fmt.Errorf("read the destroy event: %w", err)
	}
	if !foundEvent {
		envelope, err = market.PutEvent(ctx, q.repo, q.repoDID, eventRKey, market.EventOptions{
			Receipt:   market.RecordRef{URI: receipt.URI, CID: receipt.CID},
			Payload:   market.RecordRef{URI: payload.URI, CID: payload.CID},
			CreatedAt: q.nowISO(),
		}, q.signer)
		if err != nil {
			return market.SignedRecord{}, market.SignedRecord{}, fmt.Errorf("create the destroy event: %w", err)
		}
	}
	return envelope, payload, nil
}

func (r *contractRunner) runAdopt(ctx context.Context, contract v1alpha1.ComputeContract) {
	state := r.stateCopy()
	if state == nil || state.Receipt == nil {
		r.writeStatus(ctx, v1alpha1.ComputeContractFailed, "adopt: the contract status holds no bidder-signed receipt", state, nil, nil, "adopting", 1)
		return
	}
	terminated, answered := r.checkTermination(ctx, state)
	if terminated {
		r.writeStatus(ctx, v1alpha1.ComputeContractReleased, "adopt: the bidder reported the guest terminated ("+state.Termination.Reason+")", state, nil, nil, "adopted", 1)
		return
	}
	if contract.Status.Phase == v1alpha1.ComputeContractReleased {
		if answered {
			r.terminationNotReported.Store(true)
		}
		r.writeStatus(ctx, v1alpha1.ComputeContractReleased, "adopt: the contract is already released", state, nil, nil, "adopted", 1)
		return
	}
	if state.HostKeys == nil {
		r.writeStatus(ctx, v1alpha1.ComputeContractFailed, "adopt: the contract status holds no bidder-signed vm.hostKeys", state, nil, nil, "adopting", 1)
		return
	}
	if err := r.untilSettled(ctx, "adopt", func() error { return r.q.recordedWinnerAllowed(ctx, state, r.contractPolicyName()) }); err != nil {
		if r.q.stopping() {
			return
		}
		if errors.Is(err, ErrPolicyRejected) {
			state.Outcome = v1alpha1.ContractOutcomePolicyRejected
			state.Error = "adopt: " + err.Error()
			if releaseErr := r.releaseContract(ctx, state, winnerDID(state)); releaseErr != nil && !r.q.stopping() {
				r.recordFailure(ctx, state, fmt.Errorf("%s; releasing it: %w", state.Error, releaseErr), 1)
			}
			return
		}
		r.writeStatus(ctx, v1alpha1.ComputeContractFailed, "adopt: "+err.Error(), state, nil, nil, "adopting", 1)
		return
	}
	state.Phase = v1alpha1.ContractNetwork
	r.writeStatus(ctx, v1alpha1.ComputeContractHeld, "adopted: no RFP was sent and no vm.delete was submitted", state, nil, nil, "adopted", 1)
	if err := r.waitForRelease(ctx, state, state.Receipt.URI, winnerDID(state)); err != nil && !r.q.stopping() {
		r.recordFailure(ctx, r.stateCopy(), err, 1)
	}
}

func (r *contractRunner) writeStatus(ctx context.Context, phase v1alpha1.ComputeContractPhase, message string, state *v1alpha1.NodeClaimContract, bids []v1alpha1.ContractBid, chosen *v1alpha1.ContractChosen, step string, attempts int32) error {
	if r.isGone() {
		return errRunnerGone
	}
	contract := r.snapshot()
	status := v1alpha1.ComputeContractStatus{
		ObservedGeneration: contract.Generation,
		Phase:              phase,
		Message:            message,
		Bids:               append([]v1alpha1.ContractBid(nil), bids...),
		Chosen:             chosen,
		Conditions:         append([]metav1.Condition(nil), contract.Status.Conditions...),
		Execution: &v1alpha1.ExecutionStatus{
			Step:          step,
			StepStartedAt: r.q.nowISO(),
			LastEventAt:   r.q.nowISO(),
			Attempts:      attempts,
		},
		Contract:         cloneNodeClaimContract(state),
		SecretsURL:       r.secretsURLValue(),
		SecretsFetchedAt: r.secretsFetched(),
		Grant:            r.grantValue(),
	}
	if release := r.releaseStatus(); release != nil {
		status.Release = release
	} else if contract.Status.Release != nil {
		preserved := *contract.Status.Release
		status.Release = &preserved
	}
	if requested := r.releaseRequestedAt(); requested != "" && status.Release != nil &&
		(phase == v1alpha1.ComputeContractReleased || phase == v1alpha1.ComputeContractFailed) {
		status.Release.RequestedAt = requested
	}
	r.applyConditions(&status, contract.Generation, phase, message, state)
	var startRevokes func()
	if phase == v1alpha1.ComputeContractReleased || (phase == v1alpha1.ComputeContractFailed && !guestMayRun(state)) {
		r.atircRevoke.Do(func() { status.AtircRevokes, startRevokes = r.revokeAtircToken(ctx) })
	}
	err := r.q.deps.Intents.MergeComputeContractStatus(ctx, r.name, status)
	if err != nil && !notFound(err) {
		r.log().Warn("requester_contract_status_failed", "err", err)
	}
	if startRevokes != nil {
		startRevokes()
	}
	if phase == v1alpha1.ComputeContractReleased {
		r.q.dropUndelivered(receiptURIOf(state))
	}
	return err
}

var errRunnerGone = errors.New("requester: the contract runner is gone")

func notFound(err error) bool {
	return err != nil && (errors.Is(err, contractmarket.ErrNotFound) || strings.Contains(err.Error(), "not found"))
}

func (r *contractRunner) applyConditions(status *v1alpha1.ComputeContractStatus, generation int64, phase v1alpha1.ComputeContractPhase, message string, state *v1alpha1.NodeClaimContract) {
	now := r.q.now()
	addContractConditions(status, generation, phase, message, now, false)
	if r.resumed {
		conds.SetStatusAt(&status.Conditions, generation, v1alpha1.ComputeContractConditionResumed, metav1.ConditionTrue, "Recovered", "the recorded status.contract was resumed", now)
	}
	if phase == v1alpha1.ComputeContractReleasing || phase == v1alpha1.ComputeContractFailed {
		if release := r.releaseStatus(); release != nil && release.LastError != "" {
			conds.SetStatusAt(&status.Conditions, generation, v1alpha1.ComputeContractConditionReleaseFailed, metav1.ConditionTrue, "DestroyFailed", release.LastError, now)
		}
	}
	if phase == v1alpha1.ComputeContractReleased {
		acknowledged := state != nil && ((state.Release != nil && state.Release.OK) || state.Termination != nil)
		if acknowledged {
			conds.SetStatusAt(&status.Conditions, generation, v1alpha1.ComputeContractConditionReleaseAcknowledged, metav1.ConditionTrue, "Acknowledged", message, now)
			conds.SetStatusAt(&status.Conditions, generation, v1alpha1.ComputeContractConditionReleaseFailed, metav1.ConditionFalse, "Acknowledged", message, now)
		}
	}
	if state != nil && state.R2Release != nil {
		rel := state.R2Release
		ok := rel.Phase == v1alpha1.ContractR2ReleaseReleased
		message := "the guest requester's release is " + string(rel.Phase)
		if rel.Message != "" {
			message += ": " + rel.Message
		}
		conds.SetStatusAt(&status.Conditions, generation, v1alpha1.ComputeContractConditionGuestRequesterReleased, conds.Status(ok), string(rel.Phase), message, now)
	}
	switch {
	case state != nil && state.SecretsApplied != nil && state.SecretsApplied.Via == market.GuestSecretsViaSealed:
		conds.SetStatusAt(&status.Conditions, generation, v1alpha1.ComputeContractConditionSecretsApplied, metav1.ConditionTrue, "SealedFallback",
			"the guest wrote its secrets from the sealed fallback (compute.events.guest.secrets)", now)
	case state != nil && state.SecretsApplied != nil:
		conds.SetStatusAt(&status.Conditions, generation, v1alpha1.ComputeContractConditionSecretsApplied, metav1.ConditionTrue, "GetSecrets",
			"the guest wrote its secrets from the requester's getSecrets route", now)
	case r.secretsFetched() != "":
		conds.SetStatusAt(&status.Conditions, generation, v1alpha1.ComputeContractConditionSecretsApplied, metav1.ConditionTrue, "GetSecrets",
			"the guest fetched its secrets from the requester's getSecrets route", now)
	case secretsRequestPending(state):
		conds.SetStatusAt(&status.Conditions, generation, v1alpha1.ComputeContractConditionSecretsApplied, metav1.ConditionFalse, "SealedFallbackRequested",
			"the guest asked for its secrets sealed to its key; getSecrets did not answer", now)
	}
	switch {
	case state != nil && state.Termination != nil:
		conds.SetStatusAt(&status.Conditions, generation, v1alpha1.ComputeContractConditionTerminated, metav1.ConditionTrue, "BidderTerminated", terminationMessage(state.Termination), now)
	case phase == v1alpha1.ComputeContractReleased && r.terminationNotReported.Load():
		conds.SetStatusAt(&status.Conditions, generation, v1alpha1.ComputeContractConditionTerminated, metav1.ConditionFalse, market.TerminationNotReportedReason,
			"the bidder wrote no vm.terminated for the receipt by the first start after the release; it is not read again", now)
	}
	if mint := r.agentMintOutcome(); mint != nil {
		conds.SetStatusAt(&status.Conditions, generation, v1alpha1.ComputeContractConditionAtircTokenMinted, conds.Status(mint.ok), mint.reason, mint.message, now)
	}
}

func (r *contractRunner) step(ctx context.Context, step string, attempts int32) {
	state := r.stateCopy()
	r.writeStatus(ctx, r.phaseOf(state), "step "+step, state, r.currentBids(), r.currentChosen(), step, attempts)
}

func (r *contractRunner) phaseOf(state *v1alpha1.NodeClaimContract) v1alpha1.ComputeContractPhase {
	if r.q.stopping() {
		return v1alpha1.ComputeContractReleasing
	}
	if state != nil && state.Termination != nil {
		return v1alpha1.ComputeContractReleased
	}
	return v1alpha1.ComputeContractRequesting
}

func (r *contractRunner) currentBids() []v1alpha1.ContractBid {
	r.mu.Lock()
	defer r.mu.Unlock()
	return append([]v1alpha1.ContractBid(nil), r.bids...)
}

func (r *contractRunner) currentChosen() *v1alpha1.ContractChosen {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.chosen == nil {
		return nil
	}
	chosen := *r.chosen
	return &chosen
}

func (b BidRecord) arrivedAtISO() string {
	if b.ArrivedAt.IsZero() {
		return market.NowISO()
	}
	return b.ArrivedAt.UTC().Format("2006-01-02T15:04:05.000Z")
}

func cloneNodeClaimContract(in *v1alpha1.NodeClaimContract) *v1alpha1.NodeClaimContract {
	if in == nil {
		return nil
	}
	out := new(v1alpha1.NodeClaimContract)
	in.DeepCopyInto(out)
	return out
}

func winnerDID(state *v1alpha1.NodeClaimContract) string {
	if state == nil || state.Winner == nil {
		return ""
	}
	return state.Winner.DID
}

func contractHostKeys(keys []v1alpha1.ComputeVMHostKey) []v1alpha1.ContractHostKey {
	out := make([]v1alpha1.ContractHostKey, 0, len(keys))
	for _, key := range keys {
		out = append(out, v1alpha1.ContractHostKey{Type: key.Type, PublicKey: key.PublicKey, Fingerprint: key.Fingerprint})
	}
	return out
}

func (r *contractRunner) sealSecrets(ctx context.Context, state *v1alpha1.NodeClaimContract) {
	if r.released() {
		r.log().Info("requester_guest_secrets_not_sealed", "reason", "the release is requested")
		return
	}
	contract := r.snapshot()
	contract.Status.Contract = cloneNodeClaimContract(state)
	if grant := r.grantValue(); grant != nil {
		contract.Status.Grant = grant
	}
	if err := r.q.sealGuestSecrets(ctx, contract); err != nil && ctx.Err() == nil {
		r.log().Warn("requester_guest_secrets_seal_failed", "err", err)
	}
}

func payloadKind(nsid string) string {
	switch nsid {
	case market.ComputeEventsVMOnNetworkNSID:
		return "onNetwork"
	case market.ComputeEventsVMHostKeysNSID:
		return "hostKeys"
	case market.ComputeEventsVMTerminatedNSID:
		return "terminated"
	case market.ComputeEventsGuestSecretsRequestedNSID:
		return "secretsRequested"
	case market.ComputeEventsGuestSecretsAppliedNSID:
		return "secretsApplied"
	case market.ComputeEventsGuestRequesterDIDNSID:
		return "requesterDid"
	default:
		return nsid
	}
}

func defaultUserData(keys []string) string {
	var b strings.Builder
	b.WriteString("#cloud-config\n")
	if len(keys) > 0 {
		b.WriteString("ssh_authorized_keys:\n")
		for _, key := range keys {
			key = strings.TrimSpace(key)
			if key == "" {
				continue
			}
			b.WriteString("  - ")
			b.WriteString(key)
			b.WriteByte('\n')
		}
	}
	return b.String()
}

func withGuestCACerts(userData, pem string) string {
	pem = strings.TrimSpace(pem)
	if pem == "" {
		return userData
	}
	if !strings.HasPrefix(strings.TrimLeft(userData, " \t\r\n"), "#cloud-config") {
		return userData
	}
	var b strings.Builder
	b.WriteString(userData)
	if !strings.HasSuffix(userData, "\n") {
		b.WriteByte('\n')
	}
	b.WriteString("ca_certs:\n  trusted:\n    - |\n")
	for line := range strings.SplitSeq(pem, "\n") {
		line = strings.TrimSpace(line)
		if line == "" {
			continue
		}
		b.WriteString("      ")
		b.WriteString(line)
		b.WriteByte('\n')
	}
	return b.String()
}