v0.1.10: add storage-neutral cache lifecycle and task state models

core/__init__.py

@@ -3,4 +3,10 @@
 from .model_calculation import MLAmodel
 from .request import Request, Prompt, Service, Task
 from .tokenizer_registry import TokenizerRegistry
-from .fwd import forward_request
+
+
+def forward_request(*args, **kwargs):
+    """Import the optional HTTP forwarding dependency only when it is used."""
+    from .fwd import forward_request as implementation
+
+    return implementation(*args, **kwargs)

core/state_models.py

@@ -0,0 +1,156 @@
+"""Shared lifecycle contracts for cache and data-plane work."""
+
+from __future__ import annotations
+
+import hashlib
+from enum import Enum
+from typing import Any, ClassVar, Mapping
+
+from pydantic import BaseModel, ConfigDict
+
+
+class ArtifactState(str, Enum):
+    PENDING = "pending"
+    BUILDING = "building"
+    READY = "ready"
+    FAILED = "failed"
+    DELETING = "deleting"
+    DELETED = "deleted"
+
+
+class ReplicaState(str, Enum):
+    PENDING = "pending"
+    COPYING = "copying"
+    READY = "ready"
+    FAILED = "failed"
+    DELETING = "deleting"
+    DELETED = "deleted"
+
+
+class DataPlaneTaskState(str, Enum):
+    PENDING = "pending"
+    RUNNING = "running"
+    SUCCEEDED = "succeeded"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+    EXPIRED = "expired"
+
+
+class QueueWorkState(str, Enum):
+    PENDING = "pending"
+    RUNNING = "running"
+    SUCCEEDED = "succeeded"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+    SKIPPED = "skipped"
+
+
+class TransitionResult(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    previous_state: str
+    state: str
+    changed: bool
+
+
+class InvalidStateTransition(ValueError):
+    """Raised when a lifecycle transition is outside the declared contract."""
+
+    def __init__(self, current_state: Enum, target_state: Enum, allowed_targets: set[Enum]):
+        self.detail = {
+            "code": "invalid_state_transition",
+            "current_state": current_state.value,
+            "target_state": target_state.value,
+            "allowed_targets": sorted(state.value for state in allowed_targets),
+        }
+        super().__init__(str(self.detail))
+
+
+class _StateModel(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+    _transitions: ClassVar[Mapping[Enum, set[Enum]]]
+
+    def transition_to(self, target_state: Enum) -> tuple["_StateModel", TransitionResult]:
+        state = self.state
+        target = type(state)(target_state)
+        if target == state:
+            return self, TransitionResult(previous_state=state.value, state=state.value, changed=False)
+        allowed = self._transitions.get(state, set())
+        if target not in allowed:
+            raise InvalidStateTransition(state, target, allowed)
+        changed = self.model_copy(update={"state": target})
+        return changed, TransitionResult(previous_state=state.value, state=target.value, changed=True)
+
+
+ARTIFACT_TRANSITIONS = {
+    ArtifactState.PENDING: {ArtifactState.BUILDING, ArtifactState.FAILED, ArtifactState.DELETING},
+    ArtifactState.BUILDING: {ArtifactState.READY, ArtifactState.FAILED, ArtifactState.DELETING},
+    ArtifactState.READY: {ArtifactState.FAILED, ArtifactState.DELETING},
+    ArtifactState.FAILED: {ArtifactState.BUILDING, ArtifactState.DELETING},
+    ArtifactState.DELETING: {ArtifactState.DELETED, ArtifactState.FAILED},
+    ArtifactState.DELETED: set(),
+}
+
+REPLICA_TRANSITIONS = {
+    ReplicaState.PENDING: {ReplicaState.COPYING, ReplicaState.FAILED, ReplicaState.DELETING},
+    ReplicaState.COPYING: {ReplicaState.READY, ReplicaState.FAILED, ReplicaState.DELETING},
+    ReplicaState.READY: {ReplicaState.FAILED, ReplicaState.DELETING},
+    ReplicaState.FAILED: {ReplicaState.COPYING, ReplicaState.DELETING},
+    ReplicaState.DELETING: {ReplicaState.DELETED, ReplicaState.FAILED},
+    ReplicaState.DELETED: set(),
+}
+
+TASK_TRANSITIONS = {
+    DataPlaneTaskState.PENDING: {DataPlaneTaskState.RUNNING, DataPlaneTaskState.CANCELLED, DataPlaneTaskState.EXPIRED},
+    DataPlaneTaskState.RUNNING: {DataPlaneTaskState.SUCCEEDED, DataPlaneTaskState.FAILED, DataPlaneTaskState.CANCELLED, DataPlaneTaskState.EXPIRED},
+    DataPlaneTaskState.FAILED: {DataPlaneTaskState.PENDING, DataPlaneTaskState.RUNNING, DataPlaneTaskState.CANCELLED, DataPlaneTaskState.EXPIRED},
+    DataPlaneTaskState.SUCCEEDED: set(), DataPlaneTaskState.CANCELLED: set(), DataPlaneTaskState.EXPIRED: set(),
+}
+
+WORK_TRANSITIONS = {
+    QueueWorkState.PENDING: {QueueWorkState.RUNNING, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED},
+    QueueWorkState.RUNNING: {QueueWorkState.SUCCEEDED, QueueWorkState.FAILED, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED},
+    QueueWorkState.FAILED: {QueueWorkState.PENDING, QueueWorkState.RUNNING, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED},
+    QueueWorkState.SUCCEEDED: set(), QueueWorkState.CANCELLED: set(), QueueWorkState.SKIPPED: set(),
+}
+
+
+class CacheArtifact(_StateModel):
+    artifact_id: str
+    kid: str
+    state: ArtifactState = ArtifactState.PENDING
+    _transitions = ARTIFACT_TRANSITIONS
+
+
+class CacheReplica(_StateModel):
+    replica_id: str
+    artifact_id: str
+    location_key: str
+    state: ReplicaState = ReplicaState.PENDING
+    _transitions = REPLICA_TRANSITIONS
+
+
+class DataPlaneTask(_StateModel):
+    task_id: str
+    state: DataPlaneTaskState = DataPlaneTaskState.PENDING
+    _transitions = TASK_TRANSITIONS
+
+
+class QueueWork(_StateModel):
+    work_id: str
+    state: QueueWorkState = QueueWorkState.PENDING
+    _transitions = WORK_TRANSITIONS
+
+
+def stable_id(kind: str, *identity: Any) -> str:
+    """Return a deterministic identifier from an explicitly ordered identity tuple."""
+    material = "\x1f".join([kind, *(str(value) for value in identity)])
+    return f"{kind}_{hashlib.sha256(material.encode()).hexdigest()[:24]}"
+
+
+def artifact_id(kid: str) -> str:
+    return stable_id("artifact", kid)
+
+
+def replica_id(artifact: str, location_key: str) -> str:
+    return stable_id("replica", artifact, location_key)

kdn_server/legacy_state.py

@@ -0,0 +1,71 @@
+"""Read-only projection of Legacy ``kv_ready`` rows onto lifecycle models."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict
+
+from core.state_models import (
+    ArtifactState, CacheArtifact, CacheReplica, ReplicaState, artifact_id, replica_id,
+)
+
+
+class LegacyStateWarning(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+    code: str
+    message: str
+
+
+class LegacyKVStateView(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+    compatibility_status: str = "unknown"
+    artifact: CacheArtifact
+    replica: CacheReplica
+    replica_directory_exists: bool
+    warnings: tuple[LegacyStateWarning, ...] = ()
+
+
+def _safe_component(value: Any, field: str, *, allow_dot: bool = False) -> str:
+    component = str(value or "").strip()
+    if component == "." and allow_dot:
+        return component
+    path = Path(component)
+    if not component or component == ".." or (component == "." and not allow_dot) or path.is_absolute() or len(path.parts) != 1:
+        raise ValueError(f"{field} must be a non-empty single path component")
+    return component
+
+
+def map_legacy_kv_state(row: Any, kv_root: str | Path) -> LegacyKVStateView:
+    """Map a Legacy database row without mutating either storage or the filesystem."""
+    get = row.get if hasattr(row, "get") else lambda key, default=None: getattr(row, key, default)
+    kid = _safe_component(get("kid"), "kid")
+    raw_rel_dir = get("kv_rel_dir")
+    rel_dir = None
+    if raw_rel_dir is not None and str(raw_rel_dir).strip():
+        rel_dir = _safe_component(raw_rel_dir, "kv_rel_dir", allow_dot=True)
+
+    root = Path(kv_root).resolve(strict=False)
+    runtime_directory = (root / kid).resolve(strict=False)
+    if runtime_directory.parent != root:
+        raise ValueError("kid escapes the configured KV root")
+    directory_exists = runtime_directory.is_dir()
+    kv_ready = bool(get("kv_ready", False))
+    artifact_state = ArtifactState.READY if kv_ready else ArtifactState.PENDING
+    replica_state = ReplicaState.READY if kv_ready and directory_exists else ReplicaState.FAILED
+    artifact = CacheArtifact(artifact_id=artifact_id(kid), kid=kid, state=artifact_state)
+    replica = CacheReplica(
+        replica_id=replica_id(artifact.artifact_id, kid), artifact_id=artifact.artifact_id,
+        location_key=kid, state=replica_state,
+    )
+    warnings = ()
+    if rel_dir is not None and rel_dir != kid:
+        warnings = (LegacyStateWarning(
+            code="legacy_kv_rel_dir_mismatch",
+            message="Legacy kv_rel_dir differs from the runtime kid directory.",
+        ),)
+    return LegacyKVStateView(
+        artifact=artifact, replica=replica, replica_directory_exists=directory_exists,
+        warnings=warnings,
+    )

test/test_legacy_kv_state_mapping.py

@@ -0,0 +1,66 @@
+from pathlib import Path
+
+import pytest
+from pydantic import ValidationError
+
+from core.state_models import ReplicaState
+from kdn_server.legacy_state import map_legacy_kv_state
+
+
+def row(kid="kid", kv_rel_dir="kid"):
+    return {"kid": kid, "kv_ready": 1, "kv_rel_dir": kv_rel_dir}
+
+
+def warning_codes(view):
+    return {warning.code for warning in view.warnings}
+
+
+def test_runtime_kid_directory_is_healthy_and_matching_metadata_has_no_warning(tmp_path):
+    (tmp_path / "kid").mkdir()
+    view = map_legacy_kv_state(row(), tmp_path)
+    assert view.replica.state == ReplicaState.READY
+    assert view.replica_directory_exists is True
+    assert view.replica.location_key == "kid"
+    assert "legacy_kv_rel_dir_mismatch" not in warning_codes(view)
+
+
+def test_stale_existing_metadata_directory_does_not_make_replica_healthy(tmp_path):
+    (tmp_path / "stale").mkdir()
+    view = map_legacy_kv_state(row(kv_rel_dir="stale"), tmp_path)
+    assert view.replica.state == ReplicaState.FAILED
+    assert view.replica_directory_exists is False
+    assert "legacy_kv_rel_dir_mismatch" in warning_codes(view)
+
+
+def test_dot_metadata_does_not_treat_root_as_replica(tmp_path):
+    view = map_legacy_kv_state(row(kv_rel_dir="."), tmp_path)
+    assert view.replica.state == ReplicaState.FAILED
+    assert view.replica_directory_exists is False
+    assert "legacy_kv_rel_dir_mismatch" in warning_codes(view)
+
+
+def test_escaping_metadata_is_rejected(tmp_path):
+    with pytest.raises(ValueError, match="kv_rel_dir"):
+        map_legacy_kv_state(row(kv_rel_dir="../outside"), tmp_path)
+
+
+def test_projection_and_nested_models_are_frozen(tmp_path):
+    (tmp_path / "kid").mkdir()
+    view = map_legacy_kv_state(row(kv_rel_dir="other"), tmp_path)
+    with pytest.raises(ValidationError, match="Instance is frozen"):
+        view.compatibility_status = "compatible"
+    with pytest.raises(ValidationError, match="Instance is frozen"):
+        view.replica.state = ReplicaState.FAILED
+    with pytest.raises(ValidationError, match="Instance is frozen"):
+        view.warnings[0].code = "changed"
+
+
+def test_mapping_does_not_create_or_modify_files(tmp_path):
+    marker = tmp_path / "marker"
+    marker.write_text("unchanged")
+    before = {path.relative_to(tmp_path): (path.stat().st_mtime_ns, path.read_bytes())
+              for path in tmp_path.iterdir() if path.is_file()}
+    map_legacy_kv_state(row(), tmp_path)
+    after = {path.relative_to(tmp_path): (path.stat().st_mtime_ns, path.read_bytes())
+             for path in tmp_path.iterdir() if path.is_file()}
+    assert after == before

test/test_state_models.py

@@ -0,0 +1,43 @@
+import pytest
+from pydantic import ValidationError
+
+from core.state_models import (
+    ArtifactState, CacheArtifact, CacheReplica, DataPlaneTask, DataPlaneTaskState,
+    InvalidStateTransition, QueueWork, QueueWorkState, ReplicaState,
+)
+
+
+@pytest.mark.parametrize("model,target", [
+    (CacheArtifact(artifact_id="a", kid="kid"), ArtifactState.READY),
+    (CacheReplica(replica_id="r", artifact_id="a", location_key="kid"), ReplicaState.READY),
+    (DataPlaneTask(task_id="t"), DataPlaneTaskState.RUNNING),
+    (QueueWork(work_id="w"), QueueWorkState.RUNNING),
+])
+def test_lifecycle_state_is_frozen(model, target):
+    with pytest.raises(ValidationError, match="Instance is frozen"):
+        model.state = target
+
+
+def test_legal_transition_returns_changed_copy_without_mutating_source():
+    source = CacheArtifact(artifact_id="a", kid="kid")
+    building, result = source.transition_to(ArtifactState.BUILDING)
+    assert source.state == ArtifactState.PENDING
+    assert building.state == ArtifactState.BUILDING
+    assert result.changed is True
+
+
+def test_same_state_transition_is_idempotent():
+    source = DataPlaneTask(task_id="t")
+    unchanged, result = source.transition_to(DataPlaneTaskState.PENDING)
+    assert unchanged is source
+    assert result.changed is False
+
+
+def test_terminal_transition_has_structured_sorted_detail():
+    source = QueueWork(work_id="w", state=QueueWorkState.SUCCEEDED)
+    with pytest.raises(InvalidStateTransition) as caught:
+        source.transition_to(QueueWorkState.RUNNING)
+    assert caught.value.detail == {
+        "code": "invalid_state_transition", "current_state": "succeeded",
+        "target_state": "running", "allowed_targets": [],
+    }