# Mutation, sharing, phases, rotations, extension and synchronized provenance tests.
from dataclasses import dataclass, replace
import json
import pytest
from hearth import Scene, Box, Frame, Binding, ContractError
from hearth.kernel import Capability, Contract, Grant, Port, Rule, Plan, Child, Limits, SearchExhausted
from hearth.blocks import state_of
from hearth.components.primitives import Volume, Assembly, Window, Wall, Grass, Tree, Pool
from hearth.kernel.scene import Cell
from hearth.persistence import export_scene, load_scene, semantic_digest
from hearth.randomness import Scope
from hearth.composition import encapsulate, repeat, arrange, scatter
from extension_demo import Pendant, ReadingSuite


@dataclass(frozen=True)
class Editable:

    def capability(self):
        return Capability('editable', ('support',))

    def negotiate(self, c, p):
        b = Box((0, 0, 0), (0, 0, 0))
        return Contract(b, (Port('edit', 'adjust', Frame(), b, 10, 'edit'), Port('shared', 'shared-support', Frame(), b, 4)), (Grant('edit', b, ('adjust', 'excavate'), ('editor',), 1),))

    def realize(self, c, b):
        p = Plan()
        p.block((0, 0, 0), state_of('oak_log', axis='y'))
        return p


@dataclass(frozen=True)
class Editor:
    material: str
    preserve: bool = True

    def capability(self):
        return Capability('editor', ('editor',))

    def negotiate(self, c, p):
        return Contract(Box((0, 0, 0), (0, 0, 0)))

    def realize(self, c, b):
        p = Plan()
        p.block((0, 0, 0), self.material, preserve_owner=self.preserve)
        return p


def test_adjustment_membership_history_deletion_and_reload(work):
    s = Scene()
    s.place('owner', Editable())
    s.place('adjust', Editor(state_of('oak_log', axis='x')), bindings=(Binding('/owner', 'edit', 'adjust'),))
    q = s.inspect((0, 0, 0))
    assert q['instance'] == '/owner' and q['operation']['component'] == '/adjust'
    assert s.cells('/adjust') == frozenset()
    export_scene(s, work / 'adjust.litematic')
    r = load_scene(work / 'adjust.litematic')
    assert r.inspect((0, 0, 0)) == q
    s.remove('/adjust')
    assert s.inspect((0, 0, 0))['state'] == state_of('oak_log', axis='y')
    s.place('erase', Editor(state_of('air'), False), bindings=(Binding('/owner', 'edit', 'excavate'),))
    assert s.inspect((0, 0, 0))['instance'] is None and not s.cells('/owner')
    s.remove('/erase')
    assert s.cells('/owner') == {(0, 0, 0)}


def test_identical_blocks_do_not_grant_sharing_and_explicit_shared_support():
    s = Scene()
    s.place('support', Editable())
    s.place('bridge', Assembly('bridge', Box((0, 0, 0), (2, 2, 2)), ()))
    with pytest.raises(ContractError, match='write-authority'):
        s.place('intruder', Volume(Box((0, 0, 0), (0, 0, 0)), state_of('oak_log', axis='y')))
    s.share((0, 0, 0), '/bridge', '/support', 'shared')
    assert s.cells('/bridge', shared=True) == {(0, 0, 0)}
    assert ('/bridge', 'shares_support', '/support') in s.inspect((0, 0, 0))['related']
    with pytest.raises(ContractError, match='support'):
        s.remove('/support')
    s.remove('/bridge')
    assert not s.inspect((0, 0, 0))['shared']


def test_partial_tall_plant_replacement_atomicity():
    from hearth.environment import Terrain, constant
    s = Scene(domain=Box((-2, -2, -2), (4, 15, 4)))
    s.place('terrain', Terrain(s.domain, constant(0)))
    s.place('grass', Grass(True), frame=Frame((0, 1, 0)))

    @dataclass
    class Pruner:

        def capability(self):
            return Capability('pruner', ('tree',))

        def negotiate(self, c, p):
            return Contract(Box((0, 0, 0), (0, 0, 0)))

        def realize(self, c, b):
            p = Plan()
            p.block((0, 0, 0), state_of('oak_log', axis='y'))
            return p

    snapshot = (semantic_digest(s), set(s.nodes), set(s.relations), dict(s.port_uses), list(s.dependencies))
    with pytest.raises(ContractError, match='whole-object'):
        s.place('prune', Pruner(), frame=Frame((0, 1, 0)), bindings=(Binding('/grass', 'replacement', 'plant'),))
    assert snapshot == (semantic_digest(s), set(s.nodes), set(s.relations), dict(s.port_uses), list(s.dependencies))


@pytest.mark.parametrize('turn', range(4))
def test_rotated_attachment_ports_claims_membership(turn, work):
    frame = Frame((-13, 5, 17), turn)

    def compose(c):
        return Plan(children=[Child('base', Volume(Box((0, -1, 0), (8, -1, 2)), state_of('stone'))), Child('wall', Wall(thickness=3)), Child('a', Window(), frame=Frame((2, 1, 0)), bindings=(Binding(c.path + '/wall', 'infill'),)), Child('b', Window(shutters=False), frame=Frame((6, 1, 0)), bindings=(Binding(c.path + '/wall', 'infill'),))])

    s = Scene(31)
    s.place('house', encapsulate('house', Box((-1, -1, -2), (10, 8, 4)), compose), frame=frame)
    a, b = frame.point((2, 1, 0)), frame.point((6, 1, 0))
    assert s.inspect(a)['instance'] != s.inspect(b)['instance']
    assert [n['type'] for n in s.inspect(a)['chain']] == ['window.shuttered', 'wall.framed', 'house']
    assert s.nodes['/house/wall'].contract.ports[0].frame.turn == turn
    assert all(s.nodes['/house/a'].contract.envelope.contains(p) for p in s.cells('/house/a'))
    export_scene(s, work / 'rotated.litematic')
    assert load_scene(work / 'rotated.litematic').inspect(a) == s.inspect(a)


def test_completion_obligation_and_late_decoration():
    s = Scene()
    b = Box((0, 0, 0), (3, 4, 3))
    s.place('unfinished', Assembly('room', b, (), (Rule('light', ((1, 1, 1),), b, phase='complete'),)), complete=False)
    with pytest.raises(ContractError):
        s.finalize()
    s.place('lamp', Pendant(), frame=Frame((1, 2, 1)))
    s.finish_scope('/unfinished')
    s.finalize()
    with pytest.raises(ContractError, match='lighting'):
        s.remove('/lamp')


def test_extension_low_level_composite_and_nested_protocol():
    s = Scene(4)
    room = ReadingSuite()
    wrapper = encapsulate('independent-suite', Box((-2, 0, -2), (14, 11, 14)), lambda c: Plan(children=[Child('suite', room)]))
    s.place('hall', wrapper)
    s.finalize()
    assert any(n.type == 'extension.pendant' for n in s.nodes.values())
    assert any(n.type == 'extension.reading-suite' for n in s.nodes.values())


def test_independent_order_and_failed_alternative_streams():

    @dataclass
    class Seeded:

        def capability(self):
            return Capability('seeded', ())

        def negotiate(self, c, p):
            return Contract(Box((0, 0, 0), (2, 0, 0)))

        def realize(self, c, b):
            p = Plan()
            p.block((c.rng('geometry').randrange(3), 0, 0), state_of('stone'))
            return p

    a, b = Scene(81), Scene(81)
    for scene, order in ((a, ('one', 'two')), (b, ('two', 'one'))):
        for key in order:
            scene.place(key, Seeded(), frame=Frame((10 if key == 'two' else 0, 0, 0)))
    assert semantic_digest(a) == semantic_digest(b)
    before = a.cells('/one')
    with pytest.raises(SearchExhausted):
        a.choose('failed', [Child('x', Volume(Box((0, 0, 0), (2, 0, 0)), state_of('dirt')))])
    a.place('planter', Seeded(), frame=Frame((20, 0, 0)))
    assert a.cells('/one') == before
    c = Scene(81)
    c.place('one', Seeded(), seed=123)
    assert c.nodes['/one'].scope.seed == 123


def test_regeneration_restores_displaced_terrain_and_no_stale_cells():
    from hearth.environment import Terrain, constant
    s = Scene(domain=Box((-2, -3, -2), (12, 20, 12)))
    s.place('land', Terrain(s.domain, constant(0)))
    before = semantic_digest(s)
    s.place('pool', Pool(), frame=Frame((1, 0, 1)), bindings=(Binding('/land', 'construction', 'excavate'),))
    s.remove('/pool')
    assert semantic_digest(s) == before
    assert not any('/pool' in (a, c) for a, b, c in s.relations)


def test_edit_limits_unknown_space_and_budget():
    from hearth.environment import Terrain, constant
    s = Scene(domain=Box((-3, -3, -3), (12, 20, 12)))
    s.place('land', Terrain(s.domain, constant(0), edit_limit=1))
    with pytest.raises(ContractError, match='edit-limit'):
        s.place('pool', Pool(), bindings=(Binding('/land', 'construction', 'excavate'),))
    with pytest.raises(ContractError, match='unknown-space'):
        s.view.state((100, 1, 1))
    s = Scene(limits=Limits(blocks=1))
    with pytest.raises(ContractError, match='block-budget'):
        s.place('large', Volume(Box((0, 0, 0), (3, 0, 0)), state_of('stone')))
    assert not s.nodes and not s.blocks


def test_attachment_membership_across_encapsulation_boundaries():
    s = Scene()
    s.place('house', Assembly('house', Box((-2, -1, -2), (12, 8, 5)), (Child('base', Volume(Box((0, -1, 0), (8, -1, 0)), state_of('stone'))), Child('wall', Wall()))))
    s.place('facade-window', Window(), frame=Frame((3, 1, 0)), bindings=(Binding('/house/wall', 'infill'),))
    assert s.cells('/facade-window') <= s.cells('/house/wall', descendants=True)
    assert s.cells('/facade-window') <= s.cells('/house', descendants=True)
    assert s.inspect((3, 1, 0))['chain'][-1]['id'] == '/house'
    with pytest.raises(ContractError):
        s.remove('/house')
    s.regenerate('/house', dependents=('/facade-window',))
    assert s.inspect((3, 1, 0))['instance'] == '/facade-window'


def test_public_readonly_contracts_and_protected_preference():
    from hearth.components.constraints import Preserve
    from hearth.environment import Terrain, constant
    s = Scene(domain=Box((-3, -3, -3), (12, 20, 12)), preferences={'planting': constant(100)})
    s.place('land', Terrain(s.domain, constant(0)))
    s.place('reserve', Preserve(Box((1, -2, 1), (3, 1, 3))))
    assert s.view.preference('planting', (2, 0, 2)) == 100
    with pytest.raises(ContractError, match='protected-region'):
        s.place('pool', Pool(), bindings=(Binding('/land', 'construction', 'excavate'),))
    original = s.nodes['/land'].contract.decisions['water']
    public = s.view.contract('/land')
    public.decisions['water'] = 99
    assert s.nodes['/land'].contract.decisions['water'] == original


def test_explicit_candidate_budget_exhaustion():
    from hearth.kernel import Choice
    s = Scene(limits=Limits(search=1))
    s.place('occupied', Volume(Box((0, 0, 0), (0, 0, 0)), state_of('stone')))
    choice = Choice('choice', (Child('bad', Volume(Box((0, 0, 0), (0, 0, 0)), state_of('dirt'))), Child('valid', Volume(Box((3, 0, 0), (3, 0, 0)), state_of('stone')))), lambda view, path: 0)
    with pytest.raises(SearchExhausted, match='search-budget'):
        s.compose_choice(choice)
    assert '/choice' not in s.nodes
    s.limits = replace(s.limits, search=2)
    s.compose_choice(choice)
    assert s.cells('/choice') == {(3, 0, 0)}


def test_validator_extension_and_negotiated_domain():
    from hearth.kernel import validator, Diagnostic

    @validator('test.actual-stone')
    def actual_stone(scene, node, rule):
        for p in rule.cells:
            if scene.inspect(p)['state'] != 'minecraft:stone':
                yield Diagnostic('test.actual-stone', node.path, p, 'Need actual stone')

    s = Scene()
    s.place('rule', Assembly('rule', Box((0, 0, 0), (0, 0, 0)), (Child('stone', Volume(Box((0, 0, 0), (0, 0, 0)), state_of('stone'))),), (Rule('test.actual-stone', ((0, 0, 0),)),)))
    with pytest.raises(ContractError, match='test.actual-stone'):
        s.remove('/rule/stone')
    with pytest.raises(ContractError, match='input-domain'):
        s.place('bad-wall', Wall(width=100))


def test_forwarded_interfaces_discharge_after_scope_construction():
    from hearth.kernel import Choice
    later = Assembly('later', Box((0, 0, 0), (0, 0, 0)), (Child('solid', Volume(Box((0, 0, 0), (0, 0, 0)), state_of('stone'))),), ports=(Port('access', 'access', Frame(), Box((0, 0, 0), (0, 0, 0))),))
    choice = Choice('first', (Child('one', Volume(Box((0, 0, 0), (0, 0, 0)), state_of('stone'))), Child('two', Volume(Box((1, 0, 0), (1, 0, 0)), state_of('stone')))), lambda v, p: 0)
    wrapper = Assembly('wrapper', Box((0, 0, 0), (3, 1, 1)), (choice, Child('later', later, frame=Frame((3, 0, 0)))), ports=(Port('access', 'access', Frame((3, 0, 0)), Box((3, 0, 0), (3, 0, 0)), delegate=('later', 'access')),))
    s = Scene()
    s.place('root', wrapper)
    s.finalize()
    assert s.view.port('/root', 'access').frame.origin == (3, 0, 0)


def test_dense_export_budget_is_explicit(work):
    s = Scene(limits=Limits(export_volume=10))
    s.place('line', Volume(Box((0, 0, 0), (10, 0, 0)), state_of('stone')))
    with pytest.raises(ContractError, match='export-volume'):
        export_scene(s, work / 'too large.litematic')
    assert not (work / 'too large.litematic').exists()