Skip to content
Merged
11 changes: 5 additions & 6 deletions src/objectstate/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,7 @@
is_global_config_instance,
# Raw-value-preserving utilities
replace_raw,
retain_only_authored_paths,
)

# Resolver
Expand Down Expand Up @@ -100,9 +101,6 @@
register_hierarchy_relationship,
unregister_hierarchy_relationship,
get_ancestors_from_hierarchy,
# Scope key utilities
get_root_from_scope_key,
is_scope_affected,
# Context stack building (framework-agnostic)
build_context_stack,
# Threading utility for context propagation
Expand Down Expand Up @@ -181,8 +179,6 @@
'extract_all_configs',
'get_base_global_config',
'get_context_type_stack',
'get_root_from_scope_key',
'is_scope_affected',
'build_context_stack',
'spawn_thread_with_context',
# Placeholder
Expand Down Expand Up @@ -216,13 +212,14 @@
'ObjectStateEditSession',
]

__version__ = "1.3.1"
__version__ = "2.0.0"
__author__ = 'OpenHCS Team'
__description__ = 'Generic configuration framework for lazy dataclass resolution'


# Object state management
from objectstate.object_state_registry import ObjectStateRegistry
from objectstate.scope_key import ScopeKey
from objectstate.object_state import ObjectState
from objectstate.object_state_metadata import (
ObjectStateMetadataContract,
Expand Down Expand Up @@ -259,6 +256,8 @@
# Object state
'ObjectState',
'ObjectStateRegistry',
'ScopeKey',
'retain_only_authored_paths',
'ObjectStateMetadataContract',
'ObjectStateMetadataContractRegistry',
'ObjectStateMetadataStore',
Expand Down
50 changes: 0 additions & 50 deletions src/objectstate/context_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -329,56 +329,6 @@ def _is_global_type(t):
_known_hierarchy: dict = {}


def get_root_from_scope_key(scope_key: str) -> str:
"""Extract root (plate path) from scope_key for visibility checks.

scope_key format:
- Pipeline-level: just plate path (e.g., "/path/to/plate")
- Step-level: plate_path::step_token (e.g., "/path/to/plate::step_a")
- Global: empty string

Returns the portion before "::" (or the whole string if no "::" present).
"""
if not scope_key:
return ""
return scope_key.split("::")[0]


def is_scope_affected(target_scope_id: str | None, editing_scope_id: str | None) -> bool:
"""Check if target scope is affected by edit at editing scope.

Uses scope_id hierarchy - no type introspection needed:
- None/"" (global) → affects all
- "plate_path" (pipeline) → affects same plate + all its steps
- "plate_path::token" (step) → affects only that step

Args:
target_scope_id: The scope being checked for affectedness (None = global)
editing_scope_id: The scope where the edit occurred (None = global)

Returns:
True if target should refresh when editing_scope changes
"""
# Normalize None to empty string
target = target_scope_id or ""
editing = editing_scope_id or ""

# Global edit affects all
if not editing:
return True

# Different plate roots = not affected
target_root = get_root_from_scope_key(target)
editing_root = get_root_from_scope_key(editing)
if target_root != editing_root:
return False

# Same root: parent affects children, not vice versa
# editing="plate" affects target="plate::step" ✓
# editing="plate::step" affects target="plate" ✗
return target == editing or target.startswith(editing + "::")


def _normalize_type_for_hierarchy(t):
"""Normalize a type for hierarchy registry, but preserve lazy-global distinction.

Expand Down
19 changes: 0 additions & 19 deletions src/objectstate/history_migration.py

This file was deleted.

162 changes: 162 additions & 0 deletions src/objectstate/inspection.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,162 @@
"""Read-only inspection of registered ObjectStates and their fields."""

from __future__ import annotations

import logging
from collections.abc import Callable
from dataclasses import dataclass

from objectstate.object_state import ObjectState
from objectstate.object_state_registry import ObjectStateRegistry

logger = logging.getLogger(__name__)

DIRTY_MARKER = "*"
"""Marks a field (or a group of fields) whose value differs from the saved value."""

SIGNATURE_DIFF_MARKER = "_"
"""Marks a field whose raw value differs from its declared signature default."""


def field_markers(*, dirty: bool, signature_diff: bool) -> tuple[str, ...]:
"""The markers views prefix to a field, or to a group of fields, in this order."""

return tuple(
marker
for marker, present in ((DIRTY_MARKER, dirty), (SIGNATURE_DIFF_MARKER, signature_diff))
if present
)


def state_markers(state: ObjectState) -> tuple[str, ...]:
"""The markers of a whole ObjectState: any dirty field, any signature difference."""

return field_markers(
dirty=bool(state.dirty_fields),
signature_diff=bool(state.signature_diff_fields),
)


def type_name(value: object) -> str:
return "None" if value is None else type(value).__name__


def type_qualname(value_type: type | Callable[..., object]) -> str:
return f"{value_type.__module__}.{value_type.__qualname__}"


@dataclass(frozen=True, slots=True)
class FieldProvenance:
"""The ancestor scope and declaring type a field's inherited value comes from."""

source_scope_id: str | None
source_type: type
field_name: str

@property
def source_field_path(self) -> str | None:
"""The field's path in the source scope's own state, if that scope is registered."""

source_state = ObjectStateRegistry.get_by_scope(self.source_scope_id)
if source_state is None:
return None
return source_state.project_ui_visible_field_path(self.source_type, self.field_name)


@dataclass(frozen=True, slots=True)
class FieldInspection:
"""One field of one ObjectState: raw and resolved value, edit state and provenance."""

state: ObjectState
field_path: str
raw_value: object
resolved_value: object
resolved_value_available: bool
provenance: FieldProvenance | None

@classmethod
def of(cls, state: ObjectState, field_path: str):
"""Inspect ``field_path``; a value that fails to resolve is reported unavailable."""

resolved_value = None
resolved_value_available = False
try:
resolved_value = state.get_resolved_value(field_path)
resolved_value_available = True
except Exception:
logger.debug(
"Failed to resolve ObjectState field %s on scope %r.",
field_path,
state.scope_id,
exc_info=True,
)
provenance = None
match state.get_provenance(field_path):
case (source_scope_id, source_type) if source_type is not None:
provenance = FieldProvenance(
source_scope_id=source_scope_id,
source_type=source_type,
field_name=field_path.rsplit(".", 1)[-1],
)
return cls(
state=state,
field_path=field_path,
raw_value=state.parameters[field_path],
resolved_value=resolved_value,
resolved_value_available=resolved_value_available,
provenance=provenance,
)

@property
def dirty(self) -> bool:
return self.field_path in self.state.dirty_fields

@property
def signature_diff(self) -> bool:
return self.field_path in self.state.signature_diff_fields

@property
def last_changed(self) -> bool:
return self.field_path == self.state.last_changed_field

@property
def raw_value_is_none(self) -> bool:
return self.raw_value is None

@property
def resolved_value_is_none(self) -> bool:
return self.resolved_value_available and self.resolved_value is None

@property
def inherited_value(self) -> bool:
"""The field holds no value of its own and inherits a resolved one."""

return self.raw_value is None and self.resolved_value_available and (
self.resolved_value is not None
)

@property
def semantic_markers(self) -> tuple[str, ...]:
return field_markers(dirty=self.dirty, signature_diff=self.signature_diff)

@property
def field_name(self) -> str:
return self.field_path.rsplit(".", 1)[-1]

@property
def container_path(self) -> str:
return self.field_path.rsplit(".", 1)[0] if "." in self.field_path else ""

@property
def raw_value_type(self) -> str:
return type_name(self.raw_value)

@property
def resolved_value_type(self) -> str | None:
return type_name(self.resolved_value) if self.resolved_value_available else None

@property
def declaring_type(self) -> str:
"""Qualified name of the dataclass that declares this field."""

return type_qualname(self.state.type_for_path(self.field_path))
43 changes: 43 additions & 0 deletions src/objectstate/lazy_factory.py
Original file line number Diff line number Diff line change
Expand Up @@ -250,6 +250,49 @@ def replace_raw(instance, **changes):
return type(instance)(**field_values)



def retain_only_authored_paths(value, authored_field_paths) -> Any:
"""Copy a lazy dataclass, marking only ``authored_field_paths`` as explicitly set.

An ObjectState rebuild constructs every nested lazy dataclass from all its
flattened fields, so every constructor keyword reads as explicit. Clean
source generation needs the authored paths ObjectState already tracks
(its signature-diff fields); this applies them to the lazy metadata without
changing any field value.
"""

from objectstate.field_access import DottedFieldPath

paths = tuple(DottedFieldPath(path) for path in authored_field_paths)
return _copy_with_only_authored_paths(value, DottedFieldPath(""), paths)


def _copy_with_only_authored_paths(value, prefix, authored_field_paths):
from objectstate.field_access import DataclassFieldAccess

field_values = {}
for field_definition in fields(value):
field_value = DataclassFieldAccess.raw_value(value, field_definition.name)
if isinstance(field_value, LazyDataclass):
field_value = _copy_with_only_authored_paths(
field_value,
prefix.child(field_definition.name),
authored_field_paths,
)
field_values[field_definition.name] = field_value

projected = type(value)(**field_values)
object.__setattr__(
projected,
"_explicitly_set_fields",
{
field_definition.name
for field_definition in fields(projected)
if prefix.child(field_definition.name).contains_any(authored_field_paths)
},
)
return projected

# ContextEventCoordinator removed - replaced with contextvars-based context system


Expand Down
Loading
Loading