+++ /dev/null
-#!/usr/bin/env python3
-# -*- mode: python; coding: utf-8; python-indent-offset: 2 -*-
-
-"""
-ObjectRegistry
-
-Maps Python runtime objects to ProcessLocalId.
-
-Strategies:
- 1. Weak Identity (Preferred): For objects that support weakrefs (instances ,classes).
- - ID lifetime is bound to object lifetime.
- - Auto-cleaned on GC.
-
- 2. Value Identity (Fallback): For immutable primitives (int ,str ,tuple).
- - ID is bound to the *value* (hash/equality).
- - Stored strongly (since values like 42 or "red" are conceptually eternal).
-"""
-
-from __future__ import annotations
-
-import weakref
-from typing import Any ,Callable ,Dict ,Optional
-
-from .ProcessLocalId import ProcessLocalIdGenerator ,ProcessLocalId
-
-
-class ObjectRegistry:
- def __init__(self ,id_gen: ProcessLocalIdGenerator):
- self._id_gen = id_gen
-
- # Strategy 1: Entities (Weakref-able)
- self._obj_to_id_wkd: "weakref.WeakKeyDictionary[Any ,ProcessLocalId]" = weakref.WeakKeyDictionary()
- self._id_to_obj_ref: Dict[ProcessLocalId ,weakref.ref] = {}
-
- # Strategy 2: Values (Hashable ,not weakref-able)
- self._value_to_id: Dict[Any ,ProcessLocalId] = {}
-
- self._finalizers: Dict[ProcessLocalId ,Callable[[ProcessLocalId] ,None]] = {}
- self._global_finalizer: Optional[Callable[[ProcessLocalId] ,None]] = None
-
- def register_finalizer(self ,fn: Callable[[ProcessLocalId] ,None]):
- """
- Registers a finalizer callback invoked when any registered *weak-refable* object is GC'd.
- """
- self._global_finalizer = fn
-
- def _on_collect(self ,obj_id: ProcessLocalId):
- # Only called for Strategy 1 objects
- ref = self._id_to_obj_ref.pop(obj_id ,None)
- if(ref is not None):
- # The WeakKeyDictionary auto-cleans the forward mapping ,
- # but we double check or clean any edge cases if needed.
- pass
-
- fn = self._global_finalizer
- if(fn is not None):
- fn(obj_id)
-
- def get_id(self ,obj: Any) -> ProcessLocalId:
- """
- Returns the ProcessLocalId for `obj` ,registering it if needed.
- """
- # 1. Try WeakRef Strategy (Entities)
- try:
- existing = self._obj_to_id_wkd.get(obj)
- if(existing is not None):
- return existing
- except TypeError:
- # obj is not weakref-able (e.g. int ,str ,tuple ,or list).
- # Fall through to Strategy 2.
- pass
- else:
- # It IS weakref-able ,but wasn't in the dictionary. Register it.
- obj_id = self._id_gen.next_id()
- self._obj_to_id_wkd[obj] = obj_id
- # Create reverse lookup with callback
- self._id_to_obj_ref[obj_id] = weakref.ref(obj ,lambda _ref ,oid=obj_id: self._on_collect(oid))
- return obj_id
-
- # 2. Try Value Strategy (Primitives)
- # Note: Mutable non-weakrefables (like standard lists) will fail here because they are unhashable.
- try:
- existing = self._value_to_id.get(obj)
- if(existing is not None):
- return existing
-
- # Register new value
- obj_id = self._id_gen.next_id()
- self._value_to_id[obj] = obj_id
- return obj_id
- except TypeError:
- # It is neither weakref-able NOR hashable (e.g. standard list ,dict).
- raise TypeError(
- f"ObjectRegistry: cannot track object of type {type(obj)!r}. "
- "It is neither weakref-able (Entity) nor hashable (Value)."
- )
-
- def try_get_object(self ,obj_id: ProcessLocalId) -> Optional[Any]:
- """
- Best-effort: returns the live object/value.
- """
- # Check Entities
- ref = self._id_to_obj_ref.get(obj_id)
- if(ref is not None):
- return ref()
-
- # Check Values
- # For now ,we assume values identify themselves.
- for val ,pid in self._value_to_id.items():
- if(pid == obj_id):
- return val
-
- return None
+++ /dev/null
-#!/usr/bin/env python3
-# -*- mode: python; coding: utf-8; python-indent-offset: 2 -*-
-
-
-"""
-ProcessLocalId
-
-A process-local identifier used as an internal key.
-
-Design constraint:
- - NOT intended to be serialized or persisted.
- - `repr()` intentionally does not reveal the numeric token, to discourage logging/persistence.
-"""
-
-from __future__ import annotations
-
-from dataclasses import dataclass
-
-
-@dataclass(frozen=True ,slots=True)
-class ProcessLocalId:
- _n: int
-
- def __repr__(self) -> str:
- return "<ProcessLocalId>"
-
- def __str__(self) -> str:
- return "<ProcessLocalId>"
-
- def as_int_UNSAFE(self) -> int:
- """
- Returns the raw integer token.
-
- UNSAFE because:
- - tokens are process-local
- - do not write these into files/databases/logs as stable identifiers
- """
- return self._n
-
-
-class ProcessLocalIdGenerator:
- """
- Monotonic generator; ids are never recycled.
- """
- def __init__(self ,start: int = 1):
- if start < 1: raise ValueError("start must be >= 1")
- self._next_n: int = start
-
- def next_id(self) -> ProcessLocalId:
- n = self._next_n
- self._next_n += 1
- return ProcessLocalId(n)
+++ /dev/null
-#!/usr/bin/env python3
-# -*- mode: python; coding: utf-8; python-indent-offset: 2 -*-
-
-
-"""
-Property
-
-A Property is itself an entity (it has an Identity id) so that:
- - properties can have properties
- - properties can be members of semantic sets
-"""
-
-from __future__ import annotations
-
-from dataclasses import dataclass
-from typing import Optional ,Tuple
-
-from .ProcessLocalId import ProcessLocalId
-
-
-@dataclass(frozen=True ,slots=True)
-class Property:
- id: ProcessLocalId
- name_path: Tuple[str ,...]
- doc: str = ""
-
- def __repr__(self) -> str:
- # name_path is safe to reveal; id token is not.
- return f"<Property {'.'.join(self.name_path)!r}>"
+++ /dev/null
-#!/usr/bin/env python3
-# -*- mode: python; coding: utf-8; python-indent-offset: 2 -*-
-
-
-"""
-PropertyManager
-
-Core RT property system.
-
-Key decisions vs the earlier `property_manager.py`:
- - do NOT key by `object_path()` strings (avoids collisions) fileciteturn3file4
- - runtime objects are keyed by weak identity (ProcessLocalId assigned by ObjectRegistry)
- - properties are first-class entities (Property has an id), so properties can have properties
-
-This remains process-local and in-memory.
-"""
-
-from __future__ import annotations
-
-from typing import Any ,Dict ,Iterable ,List ,Optional ,Tuple ,Union
-
-from .ProcessLocalId import ProcessLocalIdGenerator ,ProcessLocalId
-from .ObjectRegistry import ObjectRegistry
-from .PropertyStore import PropertyStore
-from .Property import Property
-from .SemanticSets import SemanticSet ,SemanticSetStore
-from .Syntax import SyntaxInstance
-
-
-NamePathLike = Union[str ,List[str] ,Tuple[str ,...]]
-
-
-class PropertyManager:
- def __init__(self):
- self._id_gen = ProcessLocalIdGenerator()
- self._obj_reg = ObjectRegistry(self._id_gen)
- self._store = PropertyStore()
- self._sets = SemanticSetStore()
-
- # Declare-by-name registry
- self._name_path_to_property: Dict[Tuple[str ,...],Property] = {}
- self._property_id_to_property: Dict[ProcessLocalId ,Property] = {}
-
- self._name_path_to_set: Dict[Tuple[str ,...],SemanticSet] = {}
- self._set_id_to_set: Dict[ProcessLocalId ,SemanticSet] = {}
-
- # Optional syntax instances (if user chooses to model them)
- self._syntax_id_to_instance: Dict[ProcessLocalId ,SyntaxInstance] = {}
-
- # Finalization cleanup
- self._obj_reg.register_finalizer(self._on_subject_finalized)
-
- def _on_subject_finalized(self ,subject_id: ProcessLocalId):
- self._store.remove_subject(subject_id)
- self._sets.remove_subject(subject_id)
-
- def _normalize_name_path(self ,name_path: NamePathLike) -> Tuple[str ,...]:
- if isinstance(name_path ,tuple): return name_path
- if isinstance(name_path ,list): return tuple(name_path)
- if isinstance(name_path ,str): return tuple(name_path.split("."))
- raise TypeError("name_path must be str ,list[str] ,or tuple[str ,...]")
-
- # -------------------------
- # Identity acquisition
- # -------------------------
- def id_of_py_object(self ,obj: Any) -> ProcessLocalId:
- return self._obj_reg.get_id(obj)
-
- def create_syntax_identity(self ,syntax: SyntaxInstance) -> ProcessLocalId:
- sid = self._id_gen.next_id()
- self._syntax_id_to_instance[sid] = syntax
- return sid
-
- def try_get_syntax(self ,syntax_id: ProcessLocalId) -> Optional[SyntaxInstance]:
- return self._syntax_id_to_instance.get(syntax_id)
-
- # -------------------------
- # Property declaration
- # -------------------------
- def declare_property(self ,name_path: NamePathLike ,doc: str = "") -> ProcessLocalId:
- np = self._normalize_name_path(name_path)
- existing = self._name_path_to_property.get(np)
- if existing is not None: return existing.id
- pid = self._id_gen.next_id()
- p = Property(pid ,np ,doc)
- self._name_path_to_property[np] = p
- self._property_id_to_property[pid] = p
- return pid
-
- def property_id(self ,name_path: NamePathLike) -> ProcessLocalId:
- np = self._normalize_name_path(name_path)
- p = self._name_path_to_property.get(np)
- if p is None: raise KeyError(f"Property not declared: {np!r}")
- return p.id
-
- def try_get_property(self ,prop_id: ProcessLocalId) -> Optional[Property]:
- return self._property_id_to_property.get(prop_id)
-
- # -------------------------
- # Semantic sets
- # -------------------------
- def declare_set(self ,name_path: NamePathLike ,doc: str = "") -> ProcessLocalId:
- np = self._normalize_name_path(name_path)
- existing = self._name_path_to_set.get(np)
- if existing is not None: return existing.id
- sid = self._id_gen.next_id()
- s = SemanticSet(sid ,np ,doc)
- self._name_path_to_set[np] = s
- self._set_id_to_set[sid] = s
- return sid
-
- def add_to_set(self ,subject: Any ,set_id: ProcessLocalId):
- subject_id = self._coerce_subject_id(subject)
- self._sets.add_member(set_id ,subject_id)
-
- def is_in_set(self ,subject: Any ,set_id: ProcessLocalId) -> bool:
- subject_id = self._coerce_subject_id(subject)
- return self._sets.has_member(set_id ,subject_id)
-
- def members(self ,set_id: ProcessLocalId) -> List[ProcessLocalId]:
- return list(self._sets.members(set_id))
-
- # -------------------------
- # Set/get properties
- # -------------------------
- def set(self ,subject: Any ,prop: Union[ProcessLocalId ,NamePathLike] ,value: Any):
- subject_id = self._coerce_subject_id(subject)
- prop_id = self._coerce_property_id(prop)
- self._store.set(subject_id ,prop_id ,value)
-
- def get(self ,subject: Any ,prop: Union[ProcessLocalId ,NamePathLike] ,default: Any = None) -> Any:
- subject_id = self._coerce_subject_id(subject)
- prop_id = self._coerce_property_id(prop)
- return self._store.get(subject_id ,prop_id ,default)
-
- def has(self ,subject: Any ,prop: Union[ProcessLocalId ,NamePathLike]) -> bool:
- subject_id = self._coerce_subject_id(subject)
- prop_id = self._coerce_property_id(prop)
- return self._store.has(subject_id ,prop_id)
-
- def subjects_with(self ,prop: Union[ProcessLocalId ,NamePathLike]) -> List[ProcessLocalId]:
- prop_id = self._coerce_property_id(prop)
- return list(self._store.subjects_with(prop_id))
-
- # -------------------------
- # Coercions
- # -------------------------
- def _coerce_subject_id(self ,subject: Any) -> ProcessLocalId:
- if isinstance(subject ,ProcessLocalId): return subject
- # For Python runtime objects, we require weakref-able instances.
- return self._obj_reg.get_id(subject)
-
- def _coerce_property_id(self ,prop: Union[ProcessLocalId ,NamePathLike]) -> ProcessLocalId:
- if isinstance(prop ,ProcessLocalId): return prop
- return self.property_id(prop)
+++ /dev/null
-#!/usr/bin/env python3
-# -*- mode: python; coding: utf-8; python-indent-offset: 2 -*-
-
-
-"""
-PropertyStore
-
-Stores property values and maintains reverse lookups.
-
-This is intentionally process-local and in-memory.
-"""
-
-from __future__ import annotations
-
-from typing import Any ,Dict ,Optional ,Set ,Tuple
-
-from .ProcessLocalId import ProcessLocalId
-
-
-class PropertyStore:
- def __init__(self):
- # (subject_id ,property_id) -> value
- self._values: Dict[Tuple[ProcessLocalId ,ProcessLocalId] ,Any] = {}
-
- # subject_id -> set(property_id)
- self._subject_to_props: Dict[ProcessLocalId ,Set[ProcessLocalId]] = {}
-
- # property_id -> set(subject_id)
- self._prop_to_subjects: Dict[ProcessLocalId ,Set[ProcessLocalId]] = {}
-
- def set(self ,subject_id: ProcessLocalId ,prop_id: ProcessLocalId ,value: Any):
- key = (subject_id ,prop_id)
- self._values[key] = value
- self._subject_to_props.setdefault(subject_id ,set()).add(prop_id)
- self._prop_to_subjects.setdefault(prop_id ,set()).add(subject_id)
-
- def get(self ,subject_id: ProcessLocalId ,prop_id: ProcessLocalId ,default: Any = None) -> Any:
- return self._values.get((subject_id ,prop_id) ,default)
-
- def has(self ,subject_id: ProcessLocalId ,prop_id: ProcessLocalId) -> bool:
- return (subject_id ,prop_id) in self._values
-
- def subjects_with(self ,prop_id: ProcessLocalId) -> Set[ProcessLocalId]:
- return set(self._prop_to_subjects.get(prop_id ,set()))
-
- def props_of(self ,subject_id: ProcessLocalId) -> Set[ProcessLocalId]:
- return set(self._subject_to_props.get(subject_id ,set()))
-
- def remove_subject(self ,subject_id: ProcessLocalId):
- """
- Remove all stored properties for a subject (used on finalization).
- """
- prop_ids = self._subject_to_props.pop(subject_id ,set())
- for prop_id in prop_ids:
- self._values.pop((subject_id ,prop_id) ,None)
- s = self._prop_to_subjects.get(prop_id)
- if s is not None:
- s.discard(subject_id)
- if not s: self._prop_to_subjects.pop(prop_id ,None)
+++ /dev/null
-#!/usr/bin/env python3
-# -*- mode: python; coding: utf-8; python-indent-offset: 2 -*-
-
-
-"""
-SemanticSets
-
-Membership sets over identities. Used for semantic typing.
-
-Design:
- - set_id identifies the set
- - members are subject ids
- - reverse index for cleanup
-"""
-
-from __future__ import annotations
-
-from dataclasses import dataclass
-from typing import Dict ,Optional ,Set
-
-from .ProcessLocalId import ProcessLocalId
-
-
-@dataclass(frozen=True ,slots=True)
-class SemanticSet:
- id: ProcessLocalId
- name_path: tuple[str ,...]
- doc: str = ""
-
- def __repr__(self) -> str:
- return f"<SemanticSet {'.'.join(self.name_path)!r}>"
-
-
-class SemanticSetStore:
- def __init__(self):
- self._members: Dict[ProcessLocalId ,Set[ProcessLocalId]] = {}
- self._subject_to_sets: Dict[ProcessLocalId ,Set[ProcessLocalId]] = {}
-
- def add_member(self ,set_id: ProcessLocalId ,subject_id: ProcessLocalId):
- self._members.setdefault(set_id ,set()).add(subject_id)
- self._subject_to_sets.setdefault(subject_id ,set()).add(set_id)
-
- def has_member(self ,set_id: ProcessLocalId ,subject_id: ProcessLocalId) -> bool:
- return subject_id in self._members.get(set_id ,set())
-
- def members(self ,set_id: ProcessLocalId) -> Set[ProcessLocalId]:
- return set(self._members.get(set_id ,set()))
-
- def remove_subject(self ,subject_id: ProcessLocalId):
- set_ids = self._subject_to_sets.pop(subject_id ,set())
- for set_id in set_ids:
- m = self._members.get(set_id)
- if m is not None:
- m.discard(subject_id)
- if not m: self._members.pop(set_id ,None)
+++ /dev/null
-#!/usr/bin/env python3
-# -*- mode: python; coding: utf-8; python-indent-offset: 2 -*-
-
-
-"""
-Syntax
-
-RT syntax identity instances.
-
-We treat "syntax" as AST-level objects:
- - kind: official-ish AST node kind name (e.g., "ast.FunctionDef")
- - location: file + span
- - scope: enclosing syntax identity id (optional)
- - parts: mapping of part-name to literal or referenced syntax identity id(s)
-
-This module does NOT traverse Python programs. It only defines the data model.
-"""
-
-from __future__ import annotations
-
-from dataclasses import dataclass
-from typing import Any ,Dict ,Optional ,Tuple ,Union ,List
-
-from .ProcessLocalId import ProcessLocalId
-
-
-@dataclass(frozen=True ,slots=True)
-class SourceSpan:
- file_path: str
- lineno: int
- col: int
- end_lineno: int
- end_col: int
-
-
-SyntaxPartValue = Union[
- None
- ,bool
- ,int
- ,float
- ,str
- ,ProcessLocalId
- ,List["SyntaxPartValue"]
- ,Dict[str ,"SyntaxPartValue"]
-]
-
-
-@dataclass(frozen=True ,slots=True)
-class SyntaxInstance:
- """
- A single syntax node instance.
-
- NOTE: many syntax nodes have no identifier-name. Name-like things (identifiers)
- appear as child nodes or literals inside `parts`.
- """
- kind: str
- span: SourceSpan
- scope_id: Optional[ProcessLocalId] = None
- parts: Dict[str ,SyntaxPartValue] = None
+++ /dev/null
-from .PropertyManager import PropertyManager
--- /dev/null
+#!/usr/bin/env python3
+# -*- mode: python; coding: utf-8; python-indent-offset: 2 -*-
+
+"""
+ObjectRegistry
+
+Maps Python runtime objects to ProcessLocalId.
+
+Strategies:
+ 1. Weak Identity (Preferred): For objects that support weakrefs (instances ,classes).
+ - ID lifetime is bound to object lifetime.
+ - Auto-cleaned on GC.
+
+ 2. Value Identity (Fallback): For immutable primitives (int ,str ,tuple).
+ - ID is bound to the *value* (hash/equality).
+ - Stored strongly (since values like 42 or "red" are conceptually eternal).
+"""
+
+from __future__ import annotations
+
+import weakref
+from typing import Any ,Callable ,Dict ,Optional
+
+from .ProcessLocalId import ProcessLocalIdGenerator ,ProcessLocalId
+
+
+class ObjectRegistry:
+ def __init__(self ,id_gen: ProcessLocalIdGenerator):
+ self._id_gen = id_gen
+
+ # Strategy 1: Entities (Weakref-able)
+ self._obj_to_id_wkd: "weakref.WeakKeyDictionary[Any ,ProcessLocalId]" = weakref.WeakKeyDictionary()
+ self._id_to_obj_ref: Dict[ProcessLocalId ,weakref.ref] = {}
+
+ # Strategy 2: Values (Hashable ,not weakref-able)
+ self._value_to_id: Dict[Any ,ProcessLocalId] = {}
+
+ self._finalizers: Dict[ProcessLocalId ,Callable[[ProcessLocalId] ,None]] = {}
+ self._global_finalizer: Optional[Callable[[ProcessLocalId] ,None]] = None
+
+ def register_finalizer(self ,fn: Callable[[ProcessLocalId] ,None]):
+ """
+ Registers a finalizer callback invoked when any registered *weak-refable* object is GC'd.
+ """
+ self._global_finalizer = fn
+
+ def _on_collect(self ,obj_id: ProcessLocalId):
+ # Only called for Strategy 1 objects
+ ref = self._id_to_obj_ref.pop(obj_id ,None)
+ if(ref is not None):
+ # The WeakKeyDictionary auto-cleans the forward mapping ,
+ # but we double check or clean any edge cases if needed.
+ pass
+
+ fn = self._global_finalizer
+ if(fn is not None):
+ fn(obj_id)
+
+ def get_id(self ,obj: Any) -> ProcessLocalId:
+ """
+ Returns the ProcessLocalId for `obj` ,registering it if needed.
+ """
+ # 1. Try WeakRef Strategy (Entities)
+ try:
+ existing = self._obj_to_id_wkd.get(obj)
+ if(existing is not None):
+ return existing
+ except TypeError:
+ # obj is not weakref-able (e.g. int ,str ,tuple ,or list).
+ # Fall through to Strategy 2.
+ pass
+ else:
+ # It IS weakref-able ,but wasn't in the dictionary. Register it.
+ obj_id = self._id_gen.next_id()
+ self._obj_to_id_wkd[obj] = obj_id
+ # Create reverse lookup with callback
+ self._id_to_obj_ref[obj_id] = weakref.ref(obj ,lambda _ref ,oid=obj_id: self._on_collect(oid))
+ return obj_id
+
+ # 2. Try Value Strategy (Primitives)
+ # Note: Mutable non-weakrefables (like standard lists) will fail here because they are unhashable.
+ try:
+ existing = self._value_to_id.get(obj)
+ if(existing is not None):
+ return existing
+
+ # Register new value
+ obj_id = self._id_gen.next_id()
+ self._value_to_id[obj] = obj_id
+ return obj_id
+ except TypeError:
+ # It is neither weakref-able NOR hashable (e.g. standard list ,dict).
+ raise TypeError(
+ f"ObjectRegistry: cannot track object of type {type(obj)!r}. "
+ "It is neither weakref-able (Entity) nor hashable (Value)."
+ )
+
+ def try_get_object(self ,obj_id: ProcessLocalId) -> Optional[Any]:
+ """
+ Best-effort: returns the live object/value.
+ """
+ # Check Entities
+ ref = self._id_to_obj_ref.get(obj_id)
+ if(ref is not None):
+ return ref()
+
+ # Check Values
+ # For now ,we assume values identify themselves.
+ for val ,pid in self._value_to_id.items():
+ if(pid == obj_id):
+ return val
+
+ return None
--- /dev/null
+#!/usr/bin/env python3
+# -*- mode: python; coding: utf-8; python-indent-offset: 2 -*-
+
+
+"""
+ProcessLocalId
+
+A process-local identifier used as an internal key.
+
+Design constraint:
+ - NOT intended to be serialized or persisted.
+ - `repr()` intentionally does not reveal the numeric token, to discourage logging/persistence.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True ,slots=True)
+class ProcessLocalId:
+ _n: int
+
+ def __repr__(self) -> str:
+ return "<ProcessLocalId>"
+
+ def __str__(self) -> str:
+ return "<ProcessLocalId>"
+
+ def as_int_UNSAFE(self) -> int:
+ """
+ Returns the raw integer token.
+
+ UNSAFE because:
+ - tokens are process-local
+ - do not write these into files/databases/logs as stable identifiers
+ """
+ return self._n
+
+
+class ProcessLocalIdGenerator:
+ """
+ Monotonic generator; ids are never recycled.
+ """
+ def __init__(self ,start: int = 1):
+ if start < 1: raise ValueError("start must be >= 1")
+ self._next_n: int = start
+
+ def next_id(self) -> ProcessLocalId:
+ n = self._next_n
+ self._next_n += 1
+ return ProcessLocalId(n)
--- /dev/null
+#!/usr/bin/env python3
+# -*- mode: python; coding: utf-8; python-indent-offset: 2 -*-
+
+
+"""
+Property
+
+A Property is itself an entity (it has an Identity id) so that:
+ - properties can have properties
+ - properties can be members of semantic sets
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Optional ,Tuple
+
+from .ProcessLocalId import ProcessLocalId
+
+
+@dataclass(frozen=True ,slots=True)
+class Property:
+ id: ProcessLocalId
+ name_path: Tuple[str ,...]
+ doc: str = ""
+
+ def __repr__(self) -> str:
+ # name_path is safe to reveal; id token is not.
+ return f"<Property {'.'.join(self.name_path)!r}>"
--- /dev/null
+#!/usr/bin/env python3
+# -*- mode: python; coding: utf-8; python-indent-offset: 2 -*-
+
+
+"""
+PropertyManager
+
+Core RT property system.
+
+Key decisions vs the earlier `property_manager.py`:
+ - do NOT key by `object_path()` strings (avoids collisions) fileciteturn3file4
+ - runtime objects are keyed by weak identity (ProcessLocalId assigned by ObjectRegistry)
+ - properties are first-class entities (Property has an id), so properties can have properties
+
+This remains process-local and in-memory.
+"""
+
+from __future__ import annotations
+
+from typing import Any ,Dict ,Iterable ,List ,Optional ,Tuple ,Union
+
+from .ProcessLocalId import ProcessLocalIdGenerator ,ProcessLocalId
+from .ObjectRegistry import ObjectRegistry
+from .PropertyStore import PropertyStore
+from .Property import Property
+from .SemanticSets import SemanticSet ,SemanticSetStore
+from .Syntax import SyntaxInstance
+
+
+NamePathLike = Union[str ,List[str] ,Tuple[str ,...]]
+
+
+class PropertyManager:
+ def __init__(self):
+ self._id_gen = ProcessLocalIdGenerator()
+ self._obj_reg = ObjectRegistry(self._id_gen)
+ self._store = PropertyStore()
+ self._sets = SemanticSetStore()
+
+ # Declare-by-name registry
+ self._name_path_to_property: Dict[Tuple[str ,...],Property] = {}
+ self._property_id_to_property: Dict[ProcessLocalId ,Property] = {}
+
+ self._name_path_to_set: Dict[Tuple[str ,...],SemanticSet] = {}
+ self._set_id_to_set: Dict[ProcessLocalId ,SemanticSet] = {}
+
+ # Optional syntax instances (if user chooses to model them)
+ self._syntax_id_to_instance: Dict[ProcessLocalId ,SyntaxInstance] = {}
+
+ # Finalization cleanup
+ self._obj_reg.register_finalizer(self._on_subject_finalized)
+
+ def _on_subject_finalized(self ,subject_id: ProcessLocalId):
+ self._store.remove_subject(subject_id)
+ self._sets.remove_subject(subject_id)
+
+ def _normalize_name_path(self ,name_path: NamePathLike) -> Tuple[str ,...]:
+ if isinstance(name_path ,tuple): return name_path
+ if isinstance(name_path ,list): return tuple(name_path)
+ if isinstance(name_path ,str): return tuple(name_path.split("."))
+ raise TypeError("name_path must be str ,list[str] ,or tuple[str ,...]")
+
+ # -------------------------
+ # Identity acquisition
+ # -------------------------
+ def id_of_py_object(self ,obj: Any) -> ProcessLocalId:
+ return self._obj_reg.get_id(obj)
+
+ def create_syntax_identity(self ,syntax: SyntaxInstance) -> ProcessLocalId:
+ sid = self._id_gen.next_id()
+ self._syntax_id_to_instance[sid] = syntax
+ return sid
+
+ def try_get_syntax(self ,syntax_id: ProcessLocalId) -> Optional[SyntaxInstance]:
+ return self._syntax_id_to_instance.get(syntax_id)
+
+ # -------------------------
+ # Property declaration
+ # -------------------------
+ def declare_property(self ,name_path: NamePathLike ,doc: str = "") -> ProcessLocalId:
+ np = self._normalize_name_path(name_path)
+ existing = self._name_path_to_property.get(np)
+ if existing is not None: return existing.id
+ pid = self._id_gen.next_id()
+ p = Property(pid ,np ,doc)
+ self._name_path_to_property[np] = p
+ self._property_id_to_property[pid] = p
+ return pid
+
+ def property_id(self ,name_path: NamePathLike) -> ProcessLocalId:
+ np = self._normalize_name_path(name_path)
+ p = self._name_path_to_property.get(np)
+ if p is None: raise KeyError(f"Property not declared: {np!r}")
+ return p.id
+
+ def try_get_property(self ,prop_id: ProcessLocalId) -> Optional[Property]:
+ return self._property_id_to_property.get(prop_id)
+
+ # -------------------------
+ # Semantic sets
+ # -------------------------
+ def declare_set(self ,name_path: NamePathLike ,doc: str = "") -> ProcessLocalId:
+ np = self._normalize_name_path(name_path)
+ existing = self._name_path_to_set.get(np)
+ if existing is not None: return existing.id
+ sid = self._id_gen.next_id()
+ s = SemanticSet(sid ,np ,doc)
+ self._name_path_to_set[np] = s
+ self._set_id_to_set[sid] = s
+ return sid
+
+ def add_to_set(self ,subject: Any ,set_id: ProcessLocalId):
+ subject_id = self._coerce_subject_id(subject)
+ self._sets.add_member(set_id ,subject_id)
+
+ def is_in_set(self ,subject: Any ,set_id: ProcessLocalId) -> bool:
+ subject_id = self._coerce_subject_id(subject)
+ return self._sets.has_member(set_id ,subject_id)
+
+ def members(self ,set_id: ProcessLocalId) -> List[ProcessLocalId]:
+ return list(self._sets.members(set_id))
+
+ # -------------------------
+ # Set/get properties
+ # -------------------------
+ def set(self ,subject: Any ,prop: Union[ProcessLocalId ,NamePathLike] ,value: Any):
+ subject_id = self._coerce_subject_id(subject)
+ prop_id = self._coerce_property_id(prop)
+ self._store.set(subject_id ,prop_id ,value)
+
+ def get(self ,subject: Any ,prop: Union[ProcessLocalId ,NamePathLike] ,default: Any = None) -> Any:
+ subject_id = self._coerce_subject_id(subject)
+ prop_id = self._coerce_property_id(prop)
+ return self._store.get(subject_id ,prop_id ,default)
+
+ def has(self ,subject: Any ,prop: Union[ProcessLocalId ,NamePathLike]) -> bool:
+ subject_id = self._coerce_subject_id(subject)
+ prop_id = self._coerce_property_id(prop)
+ return self._store.has(subject_id ,prop_id)
+
+ def subjects_with(self ,prop: Union[ProcessLocalId ,NamePathLike]) -> List[ProcessLocalId]:
+ prop_id = self._coerce_property_id(prop)
+ return list(self._store.subjects_with(prop_id))
+
+ # -------------------------
+ # Coercions
+ # -------------------------
+ def _coerce_subject_id(self ,subject: Any) -> ProcessLocalId:
+ if isinstance(subject ,ProcessLocalId): return subject
+ # For Python runtime objects, we require weakref-able instances.
+ return self._obj_reg.get_id(subject)
+
+ def _coerce_property_id(self ,prop: Union[ProcessLocalId ,NamePathLike]) -> ProcessLocalId:
+ if isinstance(prop ,ProcessLocalId): return prop
+ return self.property_id(prop)
--- /dev/null
+#!/usr/bin/env python3
+# -*- mode: python; coding: utf-8; python-indent-offset: 2 -*-
+
+
+"""
+PropertyStore
+
+Stores property values and maintains reverse lookups.
+
+This is intentionally process-local and in-memory.
+"""
+
+from __future__ import annotations
+
+from typing import Any ,Dict ,Optional ,Set ,Tuple
+
+from .ProcessLocalId import ProcessLocalId
+
+
+class PropertyStore:
+ def __init__(self):
+ # (subject_id ,property_id) -> value
+ self._values: Dict[Tuple[ProcessLocalId ,ProcessLocalId] ,Any] = {}
+
+ # subject_id -> set(property_id)
+ self._subject_to_props: Dict[ProcessLocalId ,Set[ProcessLocalId]] = {}
+
+ # property_id -> set(subject_id)
+ self._prop_to_subjects: Dict[ProcessLocalId ,Set[ProcessLocalId]] = {}
+
+ def set(self ,subject_id: ProcessLocalId ,prop_id: ProcessLocalId ,value: Any):
+ key = (subject_id ,prop_id)
+ self._values[key] = value
+ self._subject_to_props.setdefault(subject_id ,set()).add(prop_id)
+ self._prop_to_subjects.setdefault(prop_id ,set()).add(subject_id)
+
+ def get(self ,subject_id: ProcessLocalId ,prop_id: ProcessLocalId ,default: Any = None) -> Any:
+ return self._values.get((subject_id ,prop_id) ,default)
+
+ def has(self ,subject_id: ProcessLocalId ,prop_id: ProcessLocalId) -> bool:
+ return (subject_id ,prop_id) in self._values
+
+ def subjects_with(self ,prop_id: ProcessLocalId) -> Set[ProcessLocalId]:
+ return set(self._prop_to_subjects.get(prop_id ,set()))
+
+ def props_of(self ,subject_id: ProcessLocalId) -> Set[ProcessLocalId]:
+ return set(self._subject_to_props.get(subject_id ,set()))
+
+ def remove_subject(self ,subject_id: ProcessLocalId):
+ """
+ Remove all stored properties for a subject (used on finalization).
+ """
+ prop_ids = self._subject_to_props.pop(subject_id ,set())
+ for prop_id in prop_ids:
+ self._values.pop((subject_id ,prop_id) ,None)
+ s = self._prop_to_subjects.get(prop_id)
+ if s is not None:
+ s.discard(subject_id)
+ if not s: self._prop_to_subjects.pop(prop_id ,None)
--- /dev/null
+#!/usr/bin/env python3
+# -*- mode: python; coding: utf-8; python-indent-offset: 2 -*-
+
+
+"""
+SemanticSets
+
+Membership sets over identities. Used for semantic typing.
+
+Design:
+ - set_id identifies the set
+ - members are subject ids
+ - reverse index for cleanup
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Dict ,Optional ,Set
+
+from .ProcessLocalId import ProcessLocalId
+
+
+@dataclass(frozen=True ,slots=True)
+class SemanticSet:
+ id: ProcessLocalId
+ name_path: tuple[str ,...]
+ doc: str = ""
+
+ def __repr__(self) -> str:
+ return f"<SemanticSet {'.'.join(self.name_path)!r}>"
+
+
+class SemanticSetStore:
+ def __init__(self):
+ self._members: Dict[ProcessLocalId ,Set[ProcessLocalId]] = {}
+ self._subject_to_sets: Dict[ProcessLocalId ,Set[ProcessLocalId]] = {}
+
+ def add_member(self ,set_id: ProcessLocalId ,subject_id: ProcessLocalId):
+ self._members.setdefault(set_id ,set()).add(subject_id)
+ self._subject_to_sets.setdefault(subject_id ,set()).add(set_id)
+
+ def has_member(self ,set_id: ProcessLocalId ,subject_id: ProcessLocalId) -> bool:
+ return subject_id in self._members.get(set_id ,set())
+
+ def members(self ,set_id: ProcessLocalId) -> Set[ProcessLocalId]:
+ return set(self._members.get(set_id ,set()))
+
+ def remove_subject(self ,subject_id: ProcessLocalId):
+ set_ids = self._subject_to_sets.pop(subject_id ,set())
+ for set_id in set_ids:
+ m = self._members.get(set_id)
+ if m is not None:
+ m.discard(subject_id)
+ if not m: self._members.pop(set_id ,None)
--- /dev/null
+#!/usr/bin/env python3
+# -*- mode: python; coding: utf-8; python-indent-offset: 2 -*-
+
+
+"""
+Syntax
+
+RT syntax identity instances.
+
+We treat "syntax" as AST-level objects:
+ - kind: official-ish AST node kind name (e.g., "ast.FunctionDef")
+ - location: file + span
+ - scope: enclosing syntax identity id (optional)
+ - parts: mapping of part-name to literal or referenced syntax identity id(s)
+
+This module does NOT traverse Python programs. It only defines the data model.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any ,Dict ,Optional ,Tuple ,Union ,List
+
+from .ProcessLocalId import ProcessLocalId
+
+
+@dataclass(frozen=True ,slots=True)
+class SourceSpan:
+ file_path: str
+ lineno: int
+ col: int
+ end_lineno: int
+ end_col: int
+
+
+SyntaxPartValue = Union[
+ None
+ ,bool
+ ,int
+ ,float
+ ,str
+ ,ProcessLocalId
+ ,List["SyntaxPartValue"]
+ ,Dict[str ,"SyntaxPartValue"]
+]
+
+
+@dataclass(frozen=True ,slots=True)
+class SyntaxInstance:
+ """
+ A single syntax node instance.
+
+ NOTE: many syntax nodes have no identifier-name. Name-like things (identifiers)
+ appear as child nodes or literals inside `parts`.
+ """
+ kind: str
+ span: SourceSpan
+ scope_id: Optional[ProcessLocalId] = None
+ parts: Dict[str ,SyntaxPartValue] = None
--- /dev/null
+from .PropertyManager import PropertyManager
projects / Epimetheus / .git / commitdiff