# Transaction kernel, automatic ownership and separate semantic/dependency graphs. import copy from dataclasses import dataclass, field, replace, is_dataclass, fields from hearth.blocks import AIR, canonical, validate_state from hearth.space import Box, Frame from hearth.randomness import Scope from .contracts import Binding, Capability, Child, Choice, Contract, Domain, Plan from .errors import ContractError, ValidationError, SearchExhausted from .view import View, ReadTrace from .validation import validate @dataclass(frozen=True) class Limits: dimension: int = 512 blocks: int = 1000000 depth: int = 40 search: int = 24 instances: int = 10000 export_volume: int = 16000000 @dataclass(frozen=True) class Cell: state: str owner: str operation: str nbt: dict | None = None shared: tuple[str, ...] = () @dataclass class Node: path: str type: str version: str parameters: dict frame: Frame scope: Scope contract: Contract parent: str | None primary: str | None = None complete: bool = True stale: bool = False bindings: tuple = () capability: Capability | None = None shared_bindings: tuple = () @dataclass(frozen=True) class Context: path: str frame: Frame scope: Scope view: View bindings: tuple[Binding, ...] limits: Limits def rng(self, purpose='plan'): return self.scope.stream(purpose) def world(self, p): return self.frame.point(p) def local(self, p): return self.frame.inverse(p) def elevation(self, x, z, maximum=None): world = self.world((x, 0, z)) cap = None if maximum is None else maximum + self.frame.origin[1] return self.view.elevation(world[0], world[2], maximum=cap) - self.frame.origin[1] def water(self, x, z): world = self.world((x, 0, z)) level = self.view.water(world[0], world[2]) return None if level is None else level - self.frame.origin[1] def state(self, point): return self.view.state(self.world(point)) class Scene: def __init__(self, seed=0, limits=None, domain=None, style=None, preferences=None): if not isinstance(seed, int): raise TypeError('Seed must be an integer') self.seed = seed self.style = dict(style or {"timber": "spruce", "infill": "minecraft:smooth_sandstone", "roof": "deepslate_tile"}) self.preferences = dict(preferences or {}) self.limits = limits or Limits() self.domain = domain self.blocks = {} self.nodes = {} self.reverse = {} self.reverse_shared = {} self.relations = set() self.members = {} self.dependencies = [] self.operations = {} self.displaced = {} self.erasures = {} self.port_uses = {} self.version = 0 self.completed = False self._recipes = {} self.events = [] @property def view(self): return View(self) def _fork(self): other = copy.copy(self) other.blocks = self.blocks.copy() other.nodes = {p: copy.copy(n) for p, n in self.nodes.items()} other.reverse = {k: set(v) for k, v in self.reverse.items()} other.reverse_shared = {k: set(v) for k, v in self.reverse_shared.items()} other.relations = set(self.relations) other.members = {k: set(v) for k, v in self.members.items()} other.dependencies = list(self.dependencies) other.operations = copy.deepcopy(self.operations) other.displaced = {k: dict(v) for k, v in self.displaced.items()} other.erasures = dict(self.erasures) other.port_uses = dict(self.port_uses) other._recipes = self._recipes.copy() other.events = list(self.events) return other def fork(self, seed=None): """Copy a frozen composition; an optional seed controls future placements only.""" result = self._fork() if seed is not None: if not isinstance(seed, int): raise TypeError('Seed must be an integer') result.seed = seed return result def _commit(self, candidate): candidate.version += 1 candidate.completed = False self.__dict__.update(candidate.__dict__) def place(self, key, component, params=None, frame=Frame(), bindings=(), seed=None, complete=True): if not isinstance(key, str) or not key or any(c in key for c in '/#'): raise ContractError('instance-key', conditions='Keys must be nonempty and exclude / and #') candidate = self._fork() path = '/' + key candidate._instantiate(path, component, params or {}, frame, tuple(bindings), None, Scope(self.seed).child(key, seed), complete) candidate._check(False) self._commit(candidate) return path def compose_choice(self, choice): trial = self._fork() trial._select_child('', choice, Frame(), Scope(self.seed)) trial._check(False) self._commit(trial) return '/' + choice.key def choose(self, key, candidates, score=lambda scene, path: 0): accepted, failed = [], [] for index, child in enumerate(candidates): if index >= self.limits.search: break trial = self._fork() try: path = trial.place(key, child.component, child.params, child.frame, child.bindings, child.seed) accepted.append((score(trial, path), index, trial, path)) except ContractError as exc: failed.append(str(exc)) if not accepted: raise SearchExhausted('search-budget', '/' + key, conditions=f'No validated candidate within budget {self.limits.search}; failures={failed}') accepted.sort(key=lambda item: (item[0], item[1])) best_score = accepted[0][0] best = [c for c in accepted if c[0] <= best_score + 0.1] chosen = Scope(self.seed).child(key).choose(best, 'alternatives') self._commit(chosen[2]) self.events.append({'kind': 'search', 'key': key, 'accepted': len(accepted), 'rejected': len(failed), 'failures': failed}) return chosen[3] def _instantiate(self, path, component, params, frame, bindings, parent, scope, complete=True): if path in self.nodes: raise ContractError('duplicate-instance', path) if len(self.nodes) >= self.limits.instances: raise ContractError('instance-budget', path) if len(path.split('/')) > self.limits.depth: raise ContractError('recursion-budget', path) cap = component.capability() for key, value in params.items(): if key not in cap.parameters or not cap.parameters[key].accepts(value): raise ContractError('parameter-domain', path, conditions=f'{key}={value!r}') for key, domain in cap.inputs.items(): if not isinstance(domain, Domain): continue value = frame if key.startswith('frame.') else component parts = key.split('.')[1:] if key.startswith('frame.') else key.split('.') for part in parts: if part.startswith('_') or not hasattr(value, part): raise ContractError('capability-input', path, conditions=key) value = getattr(value, part) if not domain.accepts(value): raise ContractError('input-domain', path, conditions=f'{key}={value!r}') trace = ReadTrace() ctx = Context(path, frame, scope, View(self, trace), bindings, self.limits) local = component.negotiate(ctx, dict(params)) contract = local.transformed(frame) if max(contract.envelope.size) > self.limits.dimension: raise ContractError('dimension-budget', path) if parent and not all(self.nodes[parent].contract.envelope.contains(p) for p in contract.envelope.corners()): raise ContractError('child-envelope', path, contract.envelope.lo, f'Exceeds accepted parent {parent}') self.nodes[path] = Node(path, cap.name, cap.version, {"construction": describe_inputs(component), "arguments": dict(params)}, frame, scope, contract, parent, complete=False, bindings=bindings, capability=cap) self.displaced[path] = {} self.reverse[path] = set() self._recipes[path] = (component, params, frame, bindings, parent, scope, complete) if parent: self.relations.add((path, 'contained_by', parent)) self.members.setdefault(parent, set()).add(path) grants = [] for binding in bindings: if binding.host not in self.nodes: raise ContractError('missing-host', path, conditions=binding.host) effective_host, port = self.view.resolved_port(binding.host, binding.port) host = self.nodes[effective_host] use_key = (binding.host, binding.port) if self.port_uses.get(use_key, 0) >= port.capacity: raise ContractError('port-capacity', path, port.frame.origin, binding.port) self.port_uses[use_key] = self.port_uses.get(use_key, 0) + 1 if binding.verb == 'install': if not port.region.contains(frame.origin): raise ContractError('port-alignment', path, frame.origin, 'Origin outside installation region') if port.frame.inverse(frame.origin)[0] % port.alignment or frame.turn % 4 != port.frame.turn % 4: raise ContractError('port-alignment', path, frame.origin, 'Outside facade bay grid') grant = next((g for g in host.contract.grants if g.key == port.grant), None) if grant: if binding.verb not in grant.verbs or (grant.tags and not set(cap.tags).intersection(grant.tags)): raise ContractError('permission-kind', path, frame.origin, f'{cap.tags} cannot {binding.verb} {binding.host}') grants.append((Binding(effective_host, port.key, binding.verb, binding.region), grant)) elif binding.verb != 'connect': raise ContractError('no-write-grant', path) kind = 'attached_to' if binding.verb == 'install' else ('connected_to' if binding.verb == 'connect' else 'supported_by') self.relations.add((path, kind, binding.host)) if kind == 'attached_to': self.members.setdefault(binding.host, set()).add(path) if kind == 'attached_to' and self.nodes[path].primary is None: self.nodes[path].primary = binding.host self.dependencies.append({'consumer': path, 'source': binding.host, 'version': self.version, 'region': binding.region or port.region.intersection(contract.envelope.expand(1)) or contract.envelope, 'reason': binding.verb}) for region in contract.reads: self.dependencies.append({'consumer': path, 'source': 'environment', 'version': self.version, 'region': region, 'reason': 'spatial-query'}) plan = component.realize(ctx, local) if trace.bounds: self.dependencies.append({'consumer': path, 'source': 'environment', 'version': self.version, 'region': trace.bounds, 'reason': 'observed-spatial-input'}) for source, region in sorted(trace.sources.items()): if source != path: self.dependencies.append({'consumer': path, 'source': source, 'version': self.version, 'region': region, 'reason': 'observed-interface-input'}) if not isinstance(plan, Plan): raise TypeError('Components return Plan effect sets') existing_consumers = set(self.nodes) - {path} final_writes = {} for write in plan.writes: p = frame.point(write.point) final_writes[p] = write protected_rules = [(n.path, r.box) for n in self.nodes.values() for r in n.contract.rules if r.kind == "protected" and r.box] retired = set() counts = {} for p, write in final_writes.items(): if not contract.envelope.contains(p): raise ContractError('envelope', path, p, 'Realization exceeds negotiated bounds') for protector, region in protected_rules: if region.contains(p) and protector != path: raise ContractError('protected-region', path, p, 'Immutable region held by ' + protector) old = self.blocks.get(p) if old and old.shared and not write.preserve_owner: raise ContractError('shared-consumer', path, p, 'Shared support requires explicit consumer rebinding') if old and write.preserve_owner and old.state.split('[')[0] != write.state.split('[')[0]: raise ContractError('property-update', path, p, 'Property adjustment must retain block type') if old and old.owner != path: authorization = next(((b, g) for b, g in grants if b.host == old.owner and g.box.contains(p) and (b.region is None or b.region.contains(p))), None) if authorization is None: raise ContractError('write-authority', path, p, f'Owned by {old.owner}; no scoped grant') binding, grant = authorization count_key = (binding.host, grant.key) counts[count_key] = counts.get(count_key, 0) + 1 if counts[count_key] > grant.limit: raise ContractError('edit-limit', path, p, f'Limit {grant.limit}') if self.nodes[old.owner].contract.atomic_object and not write.preserve_owner: if not grant.whole or not self.reverse[old.owner].issubset(final_writes): raise ContractError('whole-object', path, p, f'Partial replacement of {old.owner}') retired.add(old.owner) if old and old.owner != path: self.displaced[path].setdefault(p, old) if old: self.reverse[old.owner].discard(p) try: world_state = frame.state(write.state) validate_state(world_state) except ValueError as exc: raise ContractError('blockstate', path, p, str(exc)) from exc opid = path + '#' + write.operation operation = self.operations.setdefault(opid, {'component': path, 'name': write.operation, 'writes': 0, 'replaced': set()}) operation['writes'] += 1 if old: operation['replaced'].add(old.owner) if world_state == AIR: self.blocks.pop(p, None) self.erasures[p] = path else: self.erasures.pop(p, None) owner = old.owner if old and write.preserve_owner else path self.blocks[p] = Cell(canonical(world_state), owner, opid, write.nbt, old.shared if old and write.preserve_owner else ()) self.reverse[owner].add(p) for dependency in self.dependencies: consumer = dependency["consumer"] if consumer in existing_consumers and consumer != parent and not path.startswith(consumer + "/") and any(dependency["region"].contains(p) for p in final_writes): # Binding consumers remain valid only if their actual invariants hold. self.events.append({"kind": "revalidated-input", "consumer": consumer, "cause": path, "version": self.version}) for old in retired: self._retire(old) for child in plan.children: if isinstance(child, Choice): self._select_child(path, child, frame, scope) continue if not child.key or any(c in child.key for c in '/#'): raise ContractError('instance-key', path, conditions=child.key) self._instantiate(path + '/' + child.key, child.component, child.params, frame.compose(child.frame), child.bindings, path, scope.child(child.key, child.seed)) for port in contract.ports: if port.delegate: target, actual = self.view.resolved_port(path, port.key) if actual.frame != port.frame or actual.region != port.region or actual.kind != port.kind: raise ContractError('forwarded-interface', path, port.frame.origin, 'Realized child changed the negotiated public interface') self.relations.add((path, 'exposes', target)) for source, kind, target in plan.relations: if source not in self.nodes or target not in self.nodes: raise ContractError("dangling-relation", path, conditions=f"{source} {kind} {target}") self.relations.add((source, kind, target)) self.nodes[path].complete = complete if len(self.blocks) > self.limits.blocks: raise ContractError('block-budget', path) def _select_child(self, parent, choice, frame, scope): accepted = [] rejected = [] path = parent + '/' + choice.key for i, option in enumerate(choice.candidates[:self.limits.search]): trial = self._fork() try: trial._instantiate(path, option.component, option.params, frame.compose(option.frame), option.bindings, parent or None, scope.child(choice.key, option.seed).child(f'alternative-{i}')) trial._check(False, completed_scope=path) accepted.append((choice.score(trial.view, path), i, trial)) except ContractError as exc: rejected.append(exc.diagnostic) if not accepted: rules = {d.rule for d in rejected} if len(rules) == 1 and len(choice.candidates) <= self.limits.search: error = ContractError(next(iter(rules)), path, conditions='All declared alternatives violate the same condition: ' + str(rejected[0])) else: error = SearchExhausted('search-budget', path, conditions=str(rejected)) error.attempts = len(rejected) error.candidates = [d.__dict__ for d in rejected] raise error accepted.sort(key=lambda item: (item[0], item[1])) feasible = [a for a in accepted if a[0] <= accepted[0][0] + choice.tolerance] chosen = scope.child(choice.key).choose(feasible, 'adaptation-selection') self.__dict__.update(chosen[2].__dict__) self.events.append({'kind': 'candidate-selection', 'component': path, 'attempts': len(accepted) + len(rejected), 'accepted': len(accepted), 'rejected': [d.__dict__ for d in rejected], 'scores': [(a[1], round(a[0], 4)) for a in accepted], 'chosen': chosen[1]}) def _retire(self, path): if self.reverse.get(path): raise ContractError('whole-object', path, conditions='Cannot retire occupied instance') node = self.nodes.get(path) if node: for binding in node.bindings: key = (binding.host, binding.port) self.port_uses[key] = max(0, self.port_uses.get(key, 0) - 1) if node: for shared_host, shared_port, point in node.shared_bindings: key = (shared_host, shared_port) self.port_uses[key] = max(0, self.port_uses.get(key, 0) - 1) self.members.pop(path, None) for children in self.members.values(): children.discard(path) self.nodes.pop(path, None) self.reverse.pop(path, None) self._recipes.pop(path, None) self.relations = {r for r in self.relations if r[0] != path and r[2] != path} self.dependencies = [d for d in self.dependencies if d['consumer'] != path and d['source'] != path] self.port_uses = {k: v for k, v in self.port_uses.items() if k[0] != path} def _check(self, final, completed_scope=None): errors = validate(self, final, completed_scope) if errors: raise ValidationError(errors) def finalize(self): self._check(True) self.completed = True return self def finish_scope(self, path): trial = self._fork() trial.nodes[path].complete = True trial._check(True) self._commit(trial) def descendants(self, path): result = set() pending = [path] while pending: current = pending.pop() if current in result: continue result.add(current) pending.extend(sorted(self.members.get(current, ()))) return frozenset(result) def cells(self, path, descendants=False, shared=False): paths = [path] if descendants: paths = list(self.descendants(path)) return frozenset(p for k in paths for p in (self.reverse.get(k, set()) | (self.reverse_shared.get(k, set()) if shared else set()))) def inspect(self, point): point = tuple(point) record = self.blocks.get(point) if not record: return {'position': point, 'state': AIR, 'instance': None, 'chain': [], 'related': [], 'operation': None} chain, seen = [], set() path = record.owner while path and path not in seen: seen.add(path) node = self.nodes[path] chain.append({'id': path, 'type': node.type}) path = node.primary or node.parent related = sorted(r for r in self.relations if r[0] in seen or r[2] in seen) return {'position': point, 'state': record.state, 'instance': record.owner, 'chain': chain, 'related': related, 'operation': self.operations.get(record.operation), 'shared': record.shared, 'nbt': record.nbt} def share(self, point, participant, host, port): trial = self._fork() p = tuple(point) record = trial.blocks[p] offered = trial.view.port(host, port) if record.owner != host or offered.kind != 'shared-support' or not offered.region.contains(p) or participant not in trial.nodes: raise ContractError('shared-authority', participant, p) if participant in record.shared: return use_key = (host, port) if trial.port_uses.get(use_key, 0) >= offered.capacity: raise ContractError('port-capacity', participant, p) trial.port_uses[use_key] = trial.port_uses.get(use_key, 0) + 1 trial.reverse_shared.setdefault(participant, set()).add(p) from .contracts import Rule node = trial.nodes[participant] node.shared_bindings = node.shared_bindings + ((host, port, p),) node.contract = replace(node.contract, rules=node.contract.rules + (Rule('support', (p,)),)) trial.blocks[p] = replace(record, shared=tuple(sorted(set(record.shared + (participant,))))) trial.relations.add((participant, 'shares_support', host)) trial._check(False) self._commit(trial) def remove(self, path): trial = self._fork() trial._remove_branch(path) trial._check(False) self._commit(trial) def _remove_branch(self, path): paths = {k for k in self.nodes if k == path or k.startswith(path + '/')} affected = set() for k in sorted(paths, reverse=True): current = set(self.reverse[k]) | set(self.displaced.get(k, {})) for p in current: record = self.blocks.get(p) if record is None and self.erasures.get(p) not in paths: continue if record is not None and record.owner not in paths and self.operations[record.operation]['component'] not in paths: continue affected.add(p) if record is not None: self.reverse[record.owner].discard(p) self.erasures.pop(p, None) previous = self.displaced.get(k, {}).get(p) if previous and previous.owner in self.nodes and previous.owner not in paths: self.blocks[p] = previous self.reverse[previous.owner].add(p) else: self.blocks.pop(p, None) self.displaced.pop(k, None) for d in self.dependencies: if d['consumer'] not in paths and (d['source'] in paths or any(d['region'].contains(p) for p in affected)): self.nodes[d['consumer']].stale = True for k in sorted(paths, reverse=True): self.reverse_shared.pop(k, None) self.reverse[k].clear() self._retire(k) self.relations = {r for r in self.relations if r[0] in self.nodes and r[2] in self.nodes} for p, record in list(self.blocks.items()): if set(record.shared).intersection(paths): self.blocks[p] = replace(record, shared=tuple(s for s in record.shared if s not in paths)) def regenerate(self, path, component=None, params=None, dependents=(), seed=None): trial = self._fork() targets = [path] + list(dependents) recipes = {} for p in targets: if p in trial._recipes: recipes[p] = trial._recipes[p] elif p == path and component is not None: n = trial.nodes[p] recipes[p] = (component, n.parameters.get('arguments', {}), n.frame, n.bindings, n.parent, n.scope, n.complete) else: raise ContractError('replay-implementation', p, conditions='Reloaded branch needs an explicitly supplied component implementation') for p in targets: trial._remove_branch(p) for p in targets: comp, oldparams, frame, bindings, parent, scope, complete = recipes[p] if p == path and seed is not None: scope = scope.child('override', seed) trial._instantiate(p, component if p == path and component is not None else comp, params if p == path and params is not None else oldparams, frame, bindings, parent, scope, complete) trial._check(True) self._commit(trial) self.events.append({'kind': 'regenerated', 'targets': targets}) def describe_inputs(value): """Record declared public inputs without concrete-component dispatch.""" if value is None or isinstance(value, (str, int, float, bool)): return value if isinstance(value, (list, tuple)): return [describe_inputs(v) for v in value] if isinstance(value, dict): return {str(k): describe_inputs(v) for k, v in value.items()} if is_dataclass(value): return {'type': type(value).__module__ + '.' + type(value).__qualname__, 'inputs': {f.name: describe_inputs(getattr(value, f.name)) for f in fields(value)}} if callable(value): return {'program': value.__module__ + '.' + value.__qualname__} return {'type': type(value).__module__ + '.' + type(value).__qualname__, 'inputs': {k: describe_inputs(v) for k, v in vars(value).items() if not k.startswith('_')}}