[WIP] Model profile source normalization

sdk/nexent/core/model_service/profiles/_models_dev/inheritance.py

@@ -0,0 +1,136 @@
+"""Pinned authored base_model inheritance with per-field source ownership."""
+
+from dataclasses import dataclass
+from typing import Mapping
+
+from nexent.core.model_service.profiles._models_dev.reader import SourceDocument, SourceReadError, _freeze
+
+
+_EXCLUDED_BASE_FIELDS = frozenset({"id", "license", "links", "weights", "benchmarks"})
+
+
+@dataclass(frozen=True)
+class FieldOrigin:
+    """Source ownership before the versioned semantic mapping is applied."""
+
+    source_kind: str
+    file_path: str
+    source_path: str
+    content_sha256: str
+
+
+@dataclass(frozen=True)
+class EffectiveSourceModel:
+    """Read-only source result, distinct from a resolved Nexent outlet."""
+
+    document: SourceDocument
+    data: Mapping[str, object]
+    field_origins: Mapping[str, FieldOrigin]
+    dependencies: tuple[SourceDocument, ...]
+    omitted_paths: tuple[str, ...]
+
+
+def _error(category: str, document: SourceDocument, source_path: str) -> SourceReadError:
+    error = SourceReadError(category, document.file_path, document.kind)
+    error.source_path = source_path
+    error.rule_id = "source_inheritance_v1"
+    return error
+
+
+def _thaw(value: object) -> object:
+    if isinstance(value, Mapping):
+        return {key: _thaw(item) for key, item in value.items()}
+    if isinstance(value, (tuple, list)):
+        return [_thaw(item) for item in value]
+    return value
+
+
+def _collect_origins(value: object, path: str, document: SourceDocument, origins: dict) -> None:
+    if isinstance(value, Mapping) and value:
+        for key, item in value.items():
+            _collect_origins(item, f"{path}.{key}" if path else key, document, origins)
+    elif isinstance(value, (tuple, list)) and value:
+        for index, item in enumerate(value):
+            _collect_origins(item, f"{path}[{index}]", document, origins)
+    elif path:
+        origins[path] = FieldOrigin(document.kind, document.file_path, path, document.content_sha256)
+
+
+def _clear_origins(origins: dict, path: str) -> None:
+    for key in tuple(origins):
+        if key == path or key.startswith(path + ".") or key.startswith(path + "["):
+            del origins[key]
+
+
+def _overlay(target: dict, updates: Mapping, path: str, document: SourceDocument, origins: dict) -> None:
+    for key, value in updates.items():
+        field_path = f"{path}.{key}" if path else key
+        if isinstance(target.get(key), dict) and isinstance(value, Mapping):
+            if value:
+                origins.pop(field_path, None)
+            _overlay(target[key], value, field_path, document, origins)
+        else:
+            _clear_origins(origins, field_path)
+            target[key] = _thaw(value)
+            _collect_origins(value, field_path, document, origins)
+
+
+def _omit(target: dict, path: str, origins: dict) -> None:
+    parts = path.split(".")
+    current = target
+    parents = []
+    for part in parts[:-1]:
+        child = current.get(part)
+        if not isinstance(child, dict):
+            return
+        parents.append((current, part))
+        current = child
+    if parts[-1] not in current:
+        return
+    del current[parts[-1]]
+    _clear_origins(origins, path)
+    for parent, key in reversed(parents):
+        if parent[key]:
+            break
+        del parent[key]
+
+
+def resolve_source_model(document: SourceDocument, sources: tuple[SourceDocument, ...]) -> EffectiveSourceModel:
+    """Use only explicit base references; never infer a model from its name."""
+    if document.kind not in ("model", "provider_model"):
+        raise _error("SOURCE_MODEL_KIND_INVALID", document, "")
+    models = {}
+    for source in sources:
+        if source.kind == "model":
+            if source.key in models:
+                raise _error("SOURCE_DUPLICATE", source, "")
+            models[source.key] = source
+
+    origins = {}
+    dependencies = (document,)
+    data = {}
+    if "base_model" in document.data:
+        base_key = document.data["base_model"]
+        if document.kind != "provider_model" or not isinstance(base_key, str) or not base_key.strip():
+            raise _error("SOURCE_BASE_REFERENCE_INVALID", document, "base_model")
+        base = models.get(base_key)
+        if base is None:
+            raise _error("SOURCE_BASE_MODEL_MISSING", document, "base_model")
+        if "base_model" in base.data:
+            raise _error("SOURCE_BASE_REFERENCE_INVALID", base, "base_model")
+        data = _thaw({key: value for key, value in base.data.items() if key not in _EXCLUDED_BASE_FIELDS})
+        _collect_origins(data, "", base, origins)
+        dependencies = (base, document)
+    elif "base_model_omit" in document.data:
+        raise _error("SOURCE_OMIT_WITHOUT_BASE", document, "base_model_omit")
+
+    omissions = document.data.get("base_model_omit", ())
+    if not isinstance(omissions, (tuple, list)) or any(
+        not isinstance(path, str) or any(not part for part in path.split(".")) for path in omissions
+    ):
+        raise _error("SOURCE_OMIT_INVALID", document, "base_model_omit")
+    updates = {key: value for key, value in document.data.items() if key not in ("base_model", "base_model_omit")}
+    _overlay(data, updates, "", document, origins)
+    for path in omissions:
+        _omit(data, path, origins)
+    return EffectiveSourceModel(document, _freeze(data), _freeze(origins), dependencies, tuple(omissions))

sdk/nexent/core/model_service/profiles/_models_dev/mapping.py

@@ -0,0 +1,191 @@
+"""Versioned source dispositions, separate from runtime API field bindings."""
+
+import re
+import tomllib
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Literal, Mapping
+
+from pydantic import Field, ValidationError
+
+from nexent.core.model_service.profiles._models_dev.inheritance import EffectiveSourceModel, FieldOrigin
+from nexent.core.model_service.profiles._models_dev.reader import SourceDocument, SourceReadError, _freeze
+from nexent.core.model_service.profiles.contracts import NonBlankString, ProfileContract
+
+
+class SourceMappingRule(ProfileContract):
+    """One explicitly scoped authored field disposition."""
+
+    rule_id: NonBlankString
+    rule_version: int = Field(ge=1, le=1)
+    source_schema_ref: Literal["models_dev_authored_v1"]
+    source_kind: Literal["model", "provider", "provider_model"]
+    source_path: NonBlankString
+    source_type: Literal["boolean", "string", "integer", "number", "number_or_string", "object", "array"]
+    target_path: NonBlankString
+    transform_ref: Literal["identity_v1", "boolean_capability_v1", "zero_unknown_tokens_v1"]
+    missing_policy: Literal["omit", "unknown"]
+    special_value_policy: Literal["preserve", "zero_is_unknown", "strict_boolean"]
+    disposition: Literal["map", "reference"]
+    evidence: NonBlankString
+
+
+class SourceMappingRegistry(ProfileContract):
+    mapping_id: Literal["models_dev_source_v1"]
+    mapping_version: int = Field(ge=1, le=1)
+    rules: tuple[SourceMappingRule, ...] = Field(min_length=1)
+
+
+@dataclass(frozen=True)
+class MappedFieldOrigin:
+    source: FieldOrigin
+    rule_id: str
+    rule_version: int
+    target_path: str
+
+
+@dataclass(frozen=True)
+class MappedSourceFields:
+    """Canonical fields before supplements and runtime profile composition."""
+
+    source: EffectiveSourceModel | SourceDocument
+    mapping_id: str
+    mapping_version: int
+    values: Mapping[str, object]
+    field_origins: Mapping[str, MappedFieldOrigin]
+
+
+def _error(category: str, file_path: str, kind: str, path: str = "") -> SourceReadError:
+    error = SourceReadError(category, file_path, kind)
+    error.source_path = path
+    error.rule_id = "models_dev_source_v1"
+    return error
+
+
+def load_mapping_rules(path: Path) -> SourceMappingRegistry:
+    """Load the explicitly selected local registry; unknown rules fail closed."""
+    try:
+        data = path.read_bytes()
+    except OSError:
+        raise _error("SOURCE_MAPPING_INVALID", path.name, "mapping") from None
+    return parse_mapping_rules(data, path.name)
+
+
+def parse_mapping_rules(data: bytes, file_path: str) -> SourceMappingRegistry:
+    """Parse a captured rule snapshot; revision hashing uses these same bytes."""
+    try:
+        raw = tomllib.loads(data.decode("utf-8"))
+        if isinstance(raw.get("rules"), list):
+            raw["rules"] = tuple(raw["rules"])
+        registry = SourceMappingRegistry.model_validate(raw)
+    except (UnicodeDecodeError, tomllib.TOMLDecodeError, ValidationError):
+        raise _error("SOURCE_MAPPING_INVALID", file_path, "mapping") from None
+    identities = set()
+    rule_ids = set()
+    targets = set()
+    for rule in registry.rules:
+        identity = (rule.source_kind, rule.source_path)
+        if identity in identities or rule.rule_id in rule_ids:
+            raise _error("SOURCE_MAPPING_DUPLICATE", file_path, "mapping", rule.source_path)
+        identities.add(identity)
+        rule_ids.add(rule.rule_id)
+        target_identity = (rule.source_kind, rule.target_path)
+        if target_identity in targets:
+            raise _error("SOURCE_TARGET_CONFLICT", file_path, "mapping", rule.source_path)
+        targets.add(target_identity)
+        if (
+            not re.fullmatch(
+                r"/[a-zA-Z_][a-zA-Z0-9_]*(?:\[\])?(?:/[a-zA-Z_][a-zA-Z0-9_]*(?:\[\])?)*", rule.target_path
+            )
+            or not re.fullmatch(
+                r"[a-zA-Z_][a-zA-Z0-9_]*(?:\[\])?(?:\.[a-zA-Z_][a-zA-Z0-9_]*(?:\[\])?)*", rule.source_path
+            )
+            or rule.source_path.count("[]") != rule.target_path.count("[]")
+        ):
+            raise _error("SOURCE_MAPPING_INVALID", file_path, "mapping", rule.source_path)
+        if (rule.transform_ref == "boolean_capability_v1" and rule.source_type != "boolean") or (
+            rule.transform_ref == "zero_unknown_tokens_v1" and rule.source_type != "integer"
+        ):
+            raise _error("SOURCE_MAPPING_INVALID", file_path, "mapping", rule.source_path)
+        expected_policy = {
+            "identity_v1": "preserve",
+            "boolean_capability_v1": "strict_boolean",
+            "zero_unknown_tokens_v1": "zero_is_unknown",
+        }[rule.transform_ref]
+        if rule.special_value_policy != expected_policy:
+            raise _error("SOURCE_MAPPING_INVALID", file_path, "mapping", rule.source_path)
+    for parent in registry.rules:
+        if parent.source_type in ("object", "array"):
+            continue
+        for child in registry.rules:
+            if (
+                parent.source_kind == "provider_model"
+                and parent.source_path == "interleaved"
+                and child.source_kind == parent.source_kind
+                and child.source_path == "interleaved.field"
+            ):
+                continue
+            if child.source_kind == parent.source_kind and child.target_path.startswith(parent.target_path + "/"):
+                raise _error("SOURCE_TARGET_CONFLICT", file_path, "mapping", child.source_path)
+    return registry
+
+
+def _leaves(value: object, path: str = ""):
+    if isinstance(value, Mapping) and value:
+        for key, item in value.items():
+            yield from _leaves(item, f"{path}.{key}" if path else key)
+    elif isinstance(value, (tuple, list)) and value:
+        for index, item in enumerate(value):
+            yield from _leaves(item, f"{path}[{index}]")
+    elif path:
+        yield path, value
+
+
+def _typed(value: object, source_type: str) -> bool:
+    return {
+        "boolean": type(value) is bool,
+        "string": isinstance(value, str),
+        "integer": type(value) is int,
+        "number": type(value) in (int, float),
+        "number_or_string": type(value) in (int, float) or isinstance(value, str),
+        "object": isinstance(value, Mapping),
+        "array": isinstance(value, (tuple, list)),
+    }[source_type]
+
+
+def map_source_fields(
+    source: EffectiveSourceModel | SourceDocument, registry: SourceMappingRegistry
+) -> MappedSourceFields:
+    """Map every validated leaf explicitly; no field-name aliases or API paths."""
+    document = source.document if isinstance(source, EffectiveSourceModel) else source
+    rules = {rule.source_path: rule for rule in registry.rules if rule.source_kind == document.kind}
+    values = {}
+    origins = {}
+    for path, value in _leaves(source.data):
+        pattern = re.sub(r"\[\d+\]", "[]", path)
+        rule = rules.get(pattern)
+        if rule is None:
+            raise _error("SOURCE_FIELD_UNHANDLED", document.file_path, document.kind, path)
+        if not _typed(value, rule.source_type):
+            raise _error("SOURCE_VALUE_INVALID", document.file_path, document.kind, path)
+        indexes = iter(re.findall(r"\[(\d+)\]", path))
+        target = re.sub(r"\[\]", lambda _: "/" + next(indexes), rule.target_path)
+        if target in values:
+            raise _error("SOURCE_TARGET_CONFLICT", document.file_path, document.kind, path)
+        if rule.transform_ref == "boolean_capability_v1":
+            value = "supported" if value else "unsupported"
+        elif rule.transform_ref == "zero_unknown_tokens_v1" and value == 0:
+            value = None
+        values[target] = value
+        origins[target] = MappedFieldOrigin(
+            source.field_origins[path]
+            if isinstance(source, EffectiveSourceModel)
+            else FieldOrigin(document.kind, document.file_path, path, document.content_sha256),
+            rule.rule_id,
+            rule.rule_version,
+            target,
+        )
+    for path, rule in rules.items():
+        if "[]" not in path and rule.missing_policy == "unknown" and rule.target_path not in values:
+            values[rule.target_path] = "unknown" if rule.transform_ref == "boolean_capability_v1" else None
+    return MappedSourceFields(source, registry.mapping_id, registry.mapping_version, _freeze(values), _freeze(origins))

sdk/nexent/core/model_service/profiles/_models_dev/reader.py

@@ -92,10 +92,21 @@ def _key(entry: SourceEntry) -> str:
     raise SourceReadError("SOURCE_KIND_PATH_MISMATCH", entry.path, entry.kind)
 
 
-def read_sources(root: Path) -> tuple[SourceDocument, ...]:
+@dataclass(frozen=True)
+class SourceInventory:
+    """Metadata from the same manifest bytes used to validate source documents."""
+
+    upstream_revision: str
+    source_schema_ref: str
+    manifest_sha256: str
+    documents: tuple[SourceDocument, ...]
+
+
+def read_inventory(root: Path) -> SourceInventory:
     """Fail the whole inventory on errors; do not expose a partial catalog."""
     root = root.resolve()
-    raw_manifest = _parse(_read(root, "manifest.toml", "manifest"), "manifest.toml", "manifest")
+    manifest_bytes = _read(root, "manifest.toml", "manifest")
+    raw_manifest = _parse(manifest_bytes, "manifest.toml", "manifest")
     try:
         manifest = SourceManifest.model_validate(raw_manifest)
         if len(manifest.upstream_revision) != 40 or any(
@@ -131,4 +142,14 @@ def read_sources(root: Path) -> tuple[SourceDocument, ...]:
             raise SourceReadError("SOURCE_DIGEST_MISMATCH", entry.path, entry.kind)
         parsed = _parse(data, entry.path, entry.kind)
         documents.append(SourceDocument(entry.kind, key, entry.path, entry.content_sha256, _freeze(parsed)))
-    return tuple(documents)
+    return SourceInventory(
+        manifest.upstream_revision,
+        manifest.source_schema_ref,
+        hashlib.sha256(manifest_bytes).hexdigest(),
+        tuple(documents),
+    )
+
+
+def read_sources(root: Path) -> tuple[SourceDocument, ...]:
+    """Return complete authored documents without changing the existing interface."""
+    return read_inventory(root).documents