// SPDX-License-Identifier: MPL-2.0
#pragma once
#include "vibeqc/tensor/stream.hpp"
#include <cstring>
#include <optional>
#include <thread>

namespace vibeqc::tensor::distributed {

// CTF doi:10.1016/j.jpdc.2014.06.002 sections 4-6: global keys versus placement.
// TAMM doi:10.1063/5.0142433 III-IV/Eqs.21-24: domains and dependency completion.
// This is a bounded CPU/fake-transport capability, not either distributed runtime.
inline void require(bool ok, const char* message) {
    if (!ok) throw std::invalid_argument(message);
}
enum class Kind : std::uint16_t { domain=1, tensor, source, metric, frame, operation };
struct Id {
    std::uint64_t namespace_high=0, namespace_low=0, serial=0;
    Kind kind=Kind::domain;
    void validate(Kind expected) const {
        require((namespace_high || namespace_low) && serial && kind==expected,"invalid stable tensor ID");
    }
    bool operator==(const Id& b) const {
        return namespace_high==b.namespace_high && namespace_low==b.namespace_low && serial==b.serial && kind==b.kind;
    }
};
struct Token {
    Id id;
    std::uint64_t revision=0;
    bool operator==(const Token& b) const { return id==b.id && revision==b.revision; }
    void validate(Kind k) const { id.validate(k); require(revision>0,"missing stable revision"); }
};
// Bounded affine or selected map page, with one parent as supported by Domain.
// Larger maps/deeper dependencies explicitly refuse in this first capability.
struct DomainRef {
    Token token, parent;
    std::uint64_t parent_coordinate=0, extent=0, begin=0, step=1;
    bool dependent=false, selected=false;
    std::array<std::uint64_t,16> coordinates{};
    void validate() const {
        token.validate(Kind::domain);
        if(dependent) parent.validate(Kind::domain);
        require(extent<=SIZE_MAX && begin<=SIZE_MAX && step<=SIZE_MAX && parent_coordinate<=SIZE_MAX,"domain width overflow");
        if(selected) require(extent<=coordinates.size(),"selected domain page too large");
        else { require(step || extent<=1,"invalid domain step"); if(extent) add(begin,mul(extent-1,step)); }
        for(std::size_t i=0;selected && i<extent;++i) require(coordinates[i]<=SIZE_MAX,"coordinate overflow");
    }
    std::size_t coordinate(std::size_t i) const {
        require(i<extent,"domain coordinate out of bounds");
        return selected ? coordinates[i] : add(begin,mul(i,step));
    }
    bool operator==(const DomainRef& b) const {
        if((selected && extent>coordinates.size()) || (b.selected && b.extent>b.coordinates.size())) return false;
        if(!(token==b.token) || dependent!=b.dependent || extent!=b.extent || selected!=b.selected) return false;
        if(dependent && (!(parent==b.parent) || parent_coordinate!=b.parent_coordinate)) return false;
        if(!selected) return begin==b.begin && step==b.step;
        for(std::size_t i=0;i<extent;++i) if(coordinates[i]!=b.coordinates[i]) return false;
        return true;
    }
};
struct Descriptor {
    Id id;
    std::uint16_t schema=1;
    std::uint64_t structure_revision=1;
    DType dtype=DType::complex_fp64;
    Token source, metric, frame;
    std::size_t rank=0;
    std::array<DomainRef,8> axes{};
    void validate() const {
        id.validate(Kind::tensor); source.validate(Kind::source); metric.validate(Kind::metric); frame.validate(Kind::frame);
        require(schema==1 && structure_revision>0 && rank<=axes.size(),"unsupported tensor schema/rank");
        require(dtype==DType::fp64 || dtype==DType::complex_fp64,"unsupported tensor dtype");
        for(std::size_t i=0;i<rank;++i) axes[i].validate();
    }
    bool operator==(const Descriptor& b) const {
        if(rank>axes.size() || b.rank>b.axes.size()) return false;
        if(!(id==b.id) || schema!=b.schema || structure_revision!=b.structure_revision || dtype!=b.dtype || rank!=b.rank ||
           !(source==b.source) || !(metric==b.metric) || !(frame==b.frame)) return false;
        for(std::size_t i=0;i<rank;++i) if(!(axes[i]==b.axes[i])) return false;
        return true;
    }
};
struct Range { std::uint64_t begin=0, end=0; bool operator==(const Range& b) const { return begin==b.begin && end==b.end; } };
struct BlockKey {
    Id tensor;
    std::uint64_t structure_revision=0;
    std::size_t rank=0;
    std::array<Range,8> ranges{};
    std::array<std::uint64_t,8> parents{};
    void validate(const Descriptor& d) const {
        d.validate();
        require(tensor==d.id && structure_revision==d.structure_revision && rank==d.rank,"block identity mismatch");
        for(std::size_t i=0;i<rank;++i) {
            require(ranges[i].begin<=ranges[i].end && ranges[i].end<=d.axes[i].extent,"block range mismatch");
            require(parents[i]==(d.axes[i].dependent ? d.axes[i].parent_coordinate : 0),"ragged parent mismatch");
        }
    }
    bool operator==(const BlockKey& b) const {
        if(rank>ranges.size() || b.rank>b.ranges.size()) return false;
        if(!(tensor==b.tensor) || structure_revision!=b.structure_revision || rank!=b.rank) return false;
        for(std::size_t i=0;i<rank;++i) if(!(ranges[i]==b.ranges[i]) || parents[i]!=b.parents[i]) return false;
        return true;
    }
};
inline BlockKey full_key(const Descriptor& d) {
    d.validate(); BlockKey k; k.tensor=d.id;k.structure_revision=d.structure_revision;k.rank=d.rank;
    for(std::size_t i=0;i<d.rank;++i) { k.ranges[i]={0,d.axes[i].extent};k.parents[i]=d.axes[i].dependent ? d.axes[i].parent_coordinate : 0; }
    return k;
}
struct ValueRef {
    BlockKey key;
    std::uint64_t epoch=0;
    bool operator==(const ValueRef& b) const { return key==b.key && epoch==b.epoch; }
};

// One bounded registry generation; no eviction or reminting while bindings live.
// The method authenticates native tokens before calling authenticate/register_tensor.
class Registry {
    struct Entry { Token token; Space native; };
    struct State {
        Budget budget;
        Budget::Reservation control;
        std::array<Entry,32> tokens{};
        std::array<Descriptor,8> tensors{};
        std::size_t token_count=0,tensor_count=0;
        explicit State(const Budget& b):budget(b),control(b.reserve(sizeof(State))) {}
    };
    std::shared_ptr<State> state_;
public:
    explicit Registry(const Budget& b):state_(std::make_shared<State>(b)) {}
    static constexpr std::size_t bytes() { return sizeof(State); }
    const Budget& budget() const { return state_->budget; }
    void authenticate(Token t,Space native) {
        require(t.id.kind>=Kind::domain && t.id.kind<=Kind::operation,"invalid token kind");
        t.validate(t.id.kind);require(native.valid(),"missing authenticated native token");
        for(std::size_t i=0;i<state_->token_count;++i) {
            if(state_->tokens[i].token==t) {
                require(state_->tokens[i].native==native,"conflicting stable/native binding");return;
            }
            require(state_->tokens[i].token.id.kind!=t.id.kind || state_->tokens[i].native!=native,
                    "distinct stable identities/revisions require distinct native tokens");
        }
        require(state_->token_count<state_->tokens.size(),"binding registry full");
        state_->tokens[state_->token_count++]={t,native};
    }
    Space resolve(Token t) const {
        for(std::size_t i=0;i<state_->token_count;++i) if(state_->tokens[i].token==t) return state_->tokens[i].native;
        throw std::invalid_argument("unauthenticated stable token");
    }
    void register_tensor(const Descriptor& d) {
        d.validate();resolve(d.source);resolve(d.metric);resolve(d.frame);
        for(std::size_t i=0;i<d.rank;++i) {
            resolve(d.axes[i].token);if(d.axes[i].dependent)resolve(d.axes[i].parent);
            for(std::size_t k=0;k<i;++k) if(d.axes[k].token==d.axes[i].token &&
                d.axes[k].dependent==d.axes[i].dependent && (!d.axes[i].dependent ||
                (d.axes[k].parent==d.axes[i].parent && d.axes[k].parent_coordinate==d.axes[i].parent_coordinate)))
                require(d.axes[k]==d.axes[i],"conflicting repeated domain");
            // Domain identity/revision cannot silently change ordering between tensors.
            for(std::size_t j=0;j<state_->tensor_count;++j) for(std::size_t k=0;k<state_->tensors[j].rank;++k) {
                const auto& old=state_->tensors[j].axes[k];
                if(old.token==d.axes[i].token && old.dependent==d.axes[i].dependent &&
                   (!old.dependent || (old.parent==d.axes[i].parent && old.parent_coordinate==d.axes[i].parent_coordinate)))
                    require(old==d.axes[i],"conflicting domain definition");
            }
        }
        for(std::size_t i=0;i<state_->tensor_count;++i) if(state_->tensors[i].id==d.id) {
            require(state_->tensors[i]==d,"conflicting tensor definition");return;
        }
        require(state_->tensor_count<state_->tensors.size(),"tensor registry full");
        state_->tensors[state_->tensor_count++]=d;
    }
    void registered(const Descriptor& d) const {
        d.validate();
        for(std::size_t i=0;i<state_->tensor_count;++i) if(state_->tensors[i]==d)return;
        throw std::invalid_argument("tensor descriptor not authenticated");
    }
    SourceStamp stamp(const Descriptor& d) const { return {resolve(d.source),resolve(d.metric),resolve(d.frame),0}; }
    Domain domain(const DomainRef& d,Range range) const {
        d.validate();
        require(range.begin<=range.end && range.end<=d.extent,"invalid domain slice");
        const auto s=resolve(d.token),p=d.dependent ? resolve(d.parent) : Space{};
        if(d.selected) {
            auto temporary=budget().reserve(mul(range.end-range.begin,sizeof(std::size_t)));
            std::vector<std::size_t> coords;coords.reserve(range.end-range.begin);
            for(auto i=range.begin;i<range.end;++i)coords.push_back(d.coordinate(i));
            return Domain::selected(budget(),s,coords,p,d.parent_coordinate);
        }
        return Domain::range(s,range.end-range.begin,add(d.begin,mul(range.begin,d.step)),d.step,p,d.parent_coordinate);
    }
    class Binding {
        friend class Registry;
        Budget::Reservation control_;
        std::shared_ptr<State> registry_;
        Descriptor descriptor_;
        ValueRef value_;
        std::uint64_t placement_,version_;
        std::size_t owner_;
        View view_;
        Binding(std::shared_ptr<State> s,const Descriptor& d,ValueRef value,std::uint64_t placement,std::size_t owner,View v)
            :control_(s->budget.reserve(sizeof(Binding))),registry_(std::move(s)),descriptor_(d),value_(value),
             placement_(placement),version_(v.allocation()->version),owner_(owner),view_(std::move(v)) {}
    public:
        const Descriptor& descriptor() const { return descriptor_; }
        const ValueRef& value() const { return value_; }
        std::uint64_t placement() const { return placement_; }
        std::size_t owner() const { return owner_; }
        const View& view() const { return view_; }
        void check_owned() const {
            view_.allocation()->check_lifetime();
            require(bool(view_.allocation()->owner),"async input requires retained ownership; copy borrow first");
            require(view_.allocation()->version==version_,"stale native value binding");
        }
    };
    bool owns(const Binding& binding) const { return binding.registry_==state_; }
    std::shared_ptr<const Binding> bind(const Descriptor& d,ValueRef value,std::uint64_t placement,
                                       std::size_t owner,const View& v,SourceStamp source) const {
        registered(d);value.key.validate(d);
        require(value.epoch && placement && budget().same(v.allocation()->budget),"invalid value/placement/ledger");
        require(source==stamp(d) && v.rank()==d.rank && v.allocation()->type==d.dtype && v.access()==Access::read,"binding source/type/access mismatch");
        for(std::size_t i=0;i<d.rank;++i) {
            const auto& actual=v.axis(i).domain;const auto& expected=d.axes[i];const auto r=value.key.ranges[i];
            require(actual.space==resolve(expected.token) && actual.extent==r.end-r.begin &&
                    actual.parent==(expected.dependent ? resolve(expected.parent) : Space{}) &&
                    actual.parent_coordinate==(expected.dependent ? expected.parent_coordinate : 0),"binding domain mismatch");
            for(std::size_t j=0;j<actual.extent;++j)require(actual.coordinate(j)==expected.coordinate(r.begin+j),"binding coordinate mismatch");
        }
        return std::shared_ptr<const Binding>(new Binding(state_,d,value,placement,owner,v));
    }
};
using Binding=Registry::Binding;

struct Placement {
    std::size_t ranks=1,active=1,executor_shift=0,output_owner=0;
    std::uint64_t epoch=1;
    void validate() const { require(ranks && active && active<=ranks && executor_shift<ranks && output_owner<ranks && epoch,"invalid placement"); }
    std::size_t owner(std::size_t p) const { return p%active; }
    std::size_t executor(std::size_t p) const { return add(owner(p),executor_shift)%ranks; }
};
struct Operation {
    Token id;
    Descriptor left,right,output;
    std::uint64_t value_epoch=1;
    Placement placement;
    bool adjoint=false;
    std::size_t work_limit=0; // Scalar products plus one visit per contribution.
    void validate(const Registry& r) const {
        id.validate(Kind::operation);placement.validate();require(value_epoch>0,"missing value epoch");
        r.registered(left);r.registered(right);r.registered(output);
        require(left.rank==2 && right.rank==2 && output.rank==2 && left.dtype==DType::complex_fp64 &&
                right.dtype==left.dtype && output.dtype==left.dtype,"CPU slice requires complex matrices");
        require(left.axes[0]==right.axes[0] && left.axes[1]==output.axes[0] && right.axes[1]==output.axes[1],"contraction domain mismatch");
        require(r.stamp(left)==r.stamp(right) && r.stamp(left)==r.stamp(output),"contraction source mismatch");
        require(!(left.id==right.id) && !(left.id==output.id) && !(right.id==output.id),"distinct tensor IDs required");
        const auto work=mul(left.axes[0].extent,add(1,mul(output.axes[0].extent,output.axes[1].extent)));
        require(work<=work_limit,"declared contraction work exceeds admission");
    }
};
struct Contribution {
    Token operation;
    ValueRef output;
    std::uint64_t reduction=0;
    bool operator==(const Contribution& b) const { return operation==b.operation && output==b.output && reduction==b.reduction; }
};
struct Envelope {
    Contribution contribution;
    std::uint64_t placement_epoch=0,generation=0;
    std::size_t sender=0,receiver=0,bytes=0;
    std::uint32_t crc=0;
};
// Canonical real/imaginary binary64 network bytes. No native struct serialization.
inline void encode_double(double x,std::uint8_t* out) {
    static_assert(sizeof(double)==8 && std::numeric_limits<double>::is_iec559,"IEEE binary64 required");
    std::uint64_t bits;std::memcpy(&bits,&x,8);
    for(unsigned i=0;i<8;++i)out[i]=std::uint8_t(bits>>(56-8*i));
}
inline double decode_double(const std::uint8_t* in) {
    std::uint64_t bits=0;for(unsigned i=0;i<8;++i)bits=(bits<<8)|in[i];
    double x;std::memcpy(&x,&bits,8);return x;
}
inline std::uint32_t crc32(const std::uint8_t* p,std::size_t n) {
    std::uint32_t crc=0xffffffffu;
    for(std::size_t i=0;i<n;++i) { crc^=p[i];for(unsigned bit=0;bit<8;++bit)crc=(crc>>1)^((crc&1)?0xedb88320u:0); }
    return crc^0xffffffffu;
}

enum class Stage { ready, producing, transporting, consuming, releasing, failed, published };
class Published {
    Budget::Reservation control_;
    std::shared_ptr<Allocation> allocation_;
    ValueRef value_;
    Registry registry_;
    Descriptor descriptor_;
    friend class FakeExecutor;
    friend class CpuTransferSession;
    Published(Budget::Reservation c,std::shared_ptr<Allocation> a,ValueRef v,Registry r,Descriptor d)
        :control_(std::move(c)),allocation_(std::move(a)),value_(v),registry_(std::move(r)),descriptor_(std::move(d)) {}
public:
    Published(Published&&)=default;
    const ValueRef& value() const { return value_; }
    const Descriptor& descriptor() const { return descriptor_; }
    std::size_t size() const { return allocation_->bytes/sizeof(Complex); }
    Complex at(std::size_t i) const { require(i<size(),"published index out of range");return static_cast<const Complex*>(allocation_->base)[i]; }
};

// Single-threaded, one operation and W=1. Fake gates do not launch background
// work or touch MPI/GPU. The executor, not the user future, owns active tickets.
class FakeExecutor {
    struct State {
        Registry registry;
        Operation op;
        Budget budget;
        Budget::Reservation control,result_control,execution_allowance;
        std::shared_ptr<Allocation> partial,carry,received,sum,correction;
        std::optional<Array<std::uint8_t>> wire;
        std::shared_ptr<const Binding> left,right;
        std::optional<Lease> left_lease,right_lease;
        std::thread::id thread=std::this_thread::get_id();
        std::size_t count=0,next=0;
        std::uint64_t generation=1;
        bool busy=false,produced=false,sent=false,arrived=false,consumed=false,ack=false,future_issued=false;
        std::atomic<bool> failed{false},published{false};
        Envelope envelope;
        State(const Registry& r,Operation operation)
            :registry(r),op(std::move(operation)),budget(r.budget()),control(budget.reserve(sizeof(State))),
             result_control(budget.reserve(sizeof(Published))),execution_allowance(budget.reserve(stream_control_bytes)),
             partial(own<Complex>(budget,elements())),carry(own<Complex>(budget,elements())),
             received(own<Complex>(budget,elements())),sum(own<Complex>(budget,elements())),
             correction(own<Complex>(budget,elements())),wire(std::in_place,budget,mul(elements(),16)),count(op.left.axes[0].extent) {
            std::fill_n(static_cast<Complex*>(sum->base),elements(),Complex{});
            std::fill_n(static_cast<Complex*>(correction->base),elements(),Complex{});
        }
        std::size_t elements() const { return mul(op.output.axes[0].extent,op.output.axes[1].extent); }
        void owner() const { require(thread==std::this_thread::get_id(),"fake executor requires its owner thread"); }
        Contribution expected(std::size_t p) const { return {op.id,{full_key(op.output),op.value_epoch},p}; }
        void cancel() noexcept { if(!published) failed=true; }
        void retire() {
            if(busy && produced && sent && arrived && consumed && ack) {
                left_lease.reset();right_lease.reset();left.reset();right.reset();busy=false;
                if(generation==UINT64_MAX) { failed=true;throw std::overflow_error("ticket generation exhausted"); }
                ++generation;
                if(failed)release_workspace();
            }
        }
        void release_workspace() noexcept {
            partial.reset();carry.reset();received.reset();sum.reset();correction.reset();wire.reset();
            execution_allowance.release();result_control.release();
        }
        template<class F> void guarded(F f) {
            owner();try { require(!published,"operation already published");f(); } catch(...) { cancel();throw; }
        }
        // Temporarily replace the conservative control allowance by the actual
        // View/IndexProgram reservations, on this same ledger. No callbacks or
        // concurrent executor entry during this synchronous registered kernel.
        void produce() {
            guarded([&] {
                require(busy && !produced,"no pending producer");
                if(!failed) {
                    execution_allowance.release();
                    try {
                        std::fill_n(static_cast<Complex*>(partial->base),elements(),Complex{});
                        std::fill_n(static_cast<Complex*>(carry->base),elements(),Complex{});
                        {
                            auto a=registry.domain(op.output.axes[0],{0,op.output.axes[0].extent});
                            auto b=registry.domain(op.output.axes[1],{0,op.output.axes[1].extent});
                            View sv(partial,{{a,signed_size(b.extent)},{b,1}},0,Access::read_write);
                            View cv(carry,{{a,signed_size(b.extent)},{b,1}},0,Access::read_write);
                            CompensatedPairPlan plan({left->view(),registry.stamp(op.left)},
                                                    {right->view(),registry.stamp(op.right)},sv,cv,registry.stamp(op.left));
                            plan.execute();
                        }
                        execution_allowance=budget.reserve(stream_control_bytes);
                    } catch(...) { failed=true;throw; }
                    const auto* p=static_cast<const Complex*>(partial->base);const auto* c=static_cast<const Complex*>(carry->base);
                    for(std::size_t i=0;i<elements();++i) {
                        const auto z=finite_complex(p[i]+c[i]);encode_double(z.real(),wire->data()+16*i);encode_double(z.imag(),wire->data()+16*i+8);
                    }
                    envelope.crc=crc32(wire->data(),wire->size());
                }
                produced=true;
            });
        }
        void receive(const Envelope& e) {
            guarded([&] {
                require(busy && produced && !arrived,"receive before producer or repeated receive");
                if(!failed) {
                    require(e.contribution==envelope.contribution && e.placement_epoch==envelope.placement_epoch &&
                            e.generation==generation && e.sender==envelope.sender && e.receiver==envelope.receiver &&
                            e.bytes==wire->size() && e.crc==crc32(wire->data(),wire->size()),"invalid transfer envelope/payload");
                    auto* p=static_cast<Complex*>(received->base);
                    for(std::size_t i=0;i<elements();++i)p[i]=finite_complex({decode_double(wire->data()+16*i),decode_double(wire->data()+16*i+8)});
                }
                arrived=true;
            });
        }
        void send_done() { guarded([&] { require(busy && produced && !sent,"invalid send completion");sent=true;retire(); }); }
        void consume() {
            guarded([&] {
                require(busy && arrived && !consumed,"consumer before receive or repeated consumer");
                if(!failed) {
                    auto* s=static_cast<Complex*>(sum->base);auto* c=static_cast<Complex*>(correction->base);
                    auto* p=static_cast<const Complex*>(received->base);
                    for(std::size_t i=0;i<elements();++i)compensated_add(p[i],s[i],c[i]);
                    ++next;
                }
                consumed=true;
            });
        }
        void receipt(std::uint64_t g) {
            guarded([&] { require(busy && consumed && !ack && g==generation,"invalid consumer receipt");ack=true;retire(); });
        }
        // Fake-only destruction can synchronously settle all simulated activity.
        // This is NOT a cancellation/drain algorithm for real MPI or device work.
        void drain() noexcept {
            cancel();left_lease.reset();right_lease.reset();left.reset();right.reset();busy=false;release_workspace();
        }
    };
    std::shared_ptr<State> state_;
public:
    // Exact reserved census, before provider reads; caller inventories/registry/
    // bindings already live on the same Budget and are not charged twice.
    static std::size_t admission_bytes(const Registry& r,const Operation& op) {
        op.validate(r);const auto n=mul(op.output.axes[0].extent,op.output.axes[1].extent);
        return add(add(sizeof(State),sizeof(Published)),add(stream_control_bytes,add(mul(5,sizeof(Allocation)),mul(n,96))));
    }
    FakeExecutor(const Registry& r,const Operation& op) {
        const auto bytes=admission_bytes(r,op);
        require(r.budget().available()>=bytes,"fake operation budget refused before input reads");
        state_=std::make_shared<State>(r,op);
    }
    FakeExecutor(const FakeExecutor&)=delete;
    FakeExecutor& operator=(const FakeExecutor&)=delete;
    ~FakeExecutor() { if(state_ && !state_->published)state_->drain(); }
    class Future {
        std::shared_ptr<State> state_;
        friend class FakeExecutor;
        explicit Future(std::shared_ptr<State> s):state_(std::move(s)) {}
    public:
        Future(const Future&)=delete;
        Future(Future&&)=default;
        ~Future() { if(state_)state_->cancel(); }
        bool failed() const { return state_->failed; }
    };
    Future future() {
        state_->owner();require(!state_->future_issued,"one future per operation");
        state_->future_issued=true;return Future(state_);
    }
    Stage stage() const {
        const auto& s=*state_;s.owner();
        if(s.failed)return Stage::failed;if(s.published)return Stage::published;if(!s.busy)return Stage::ready;
        if(!s.produced)return Stage::producing;if(!s.arrived)return Stage::transporting;
        if(!s.consumed)return Stage::consuming;return Stage::releasing;
    }
    std::size_t accepted() const { state_->owner();return state_->next; }
    bool pending() const { state_->owner();return state_->busy; }
    Contribution expected(std::size_t p) const { state_->owner();return state_->expected(p); }
    void submit(const Contribution& id,std::shared_ptr<const Binding> left,std::shared_ptr<const Binding> right,std::size_t executor) {
        auto& s=*state_;s.guarded([&] {
            require(!s.failed && !s.busy && s.next<s.count && id==s.expected(s.next),"duplicate/missing contribution or no request credit");
            require(left && right,"missing tensor binding");
            for(const auto& pair:{std::make_pair(left,&s.op.left),std::make_pair(right,&s.op.right)}) {
                auto key=full_key(*pair.second);key.ranges[0]={s.next,s.next+1};
                require(pair.first->descriptor()==*pair.second && pair.first->value()==ValueRef{key,s.op.value_epoch} &&
                        pair.first->placement()==s.op.placement.epoch && pair.first->owner()==s.op.placement.owner(s.next),"stale or misplaced input");
                require(s.registry.owns(*pair.first) && s.budget.same(pair.first->view().allocation()->budget),"foreign input ledger");
                pair.first->check_owned();
            }
            require(executor==s.op.placement.executor(s.next) && left->view().conjugate()==s.op.adjoint && !right->view().conjugate(),"executor/conjugation mismatch");
            // Acquire both before installing the ticket; failure releases the first.
            s.left_lease.emplace(left->view().allocation(),false);
            try { s.right_lease.emplace(right->view().allocation(),false); } catch(...) { s.left_lease.reset();throw; }
            s.left=std::move(left);s.right=std::move(right);
            s.busy=true;s.produced=s.sent=s.arrived=s.consumed=s.ack=false;
            s.envelope={id,s.op.placement.epoch,s.generation,executor,s.op.placement.output_owner,mul(s.elements(),16),0};
        });
    }
    void producer_ready() { state_->produce(); }
    Envelope envelope() const { state_->owner();require(state_->busy && state_->produced,"payload not ready");return state_->envelope; }
    void receive_complete(const Envelope& e) { state_->receive(e); }
    void send_complete() { state_->send_done(); }
    void consumer_complete() { state_->consume(); }
    void release_receipt(std::uint64_t generation) { state_->receipt(generation); }
    void fail_provider() { state_->owner();state_->cancel(); }
    void drain_fake() { state_->owner();state_->drain(); }
    // Explicit fault seam for fake transport only, bounded and never exposed to MPI.
    void corrupt_byte(std::size_t i) { state_->owner();require(state_->busy && state_->produced && i<state_->wire->size(),"invalid fault byte");(*state_->wire)[i]^=1; }
    Published publish() {
        auto& s=*state_;s.owner();
        try {
            require(!s.failed && !s.published && !s.busy && s.next==s.count,"incomplete or failed publication");
            auto* p=static_cast<Complex*>(s.sum->base);auto* c=static_cast<const Complex*>(s.correction->base);
            for(std::size_t i=0;i<s.elements();++i)p[i]=finite_complex(p[i]+c[i]);
            s.sum->writable=false;s.published=true;
            Published result(std::move(s.result_control),std::move(s.sum),{full_key(s.op.output),s.op.value_epoch},s.registry,s.op.output);
            s.release_workspace();
            return result;
        } catch(...) { s.cancel();throw; }
    }
};
} // namespace vibeqc::tensor::distributed