diff --git a/README.md b/README.md index 8c384fd..9d8284d 100644 --- a/README.md +++ b/README.md @@ -647,6 +647,7 @@ curl -s http://127.0.0.1:7001/debug/strategy | Document | Description | |---|---| +| [`doc/state-models.md`](doc/state-models.md) | Immutable cache lifecycle, task states, stable IDs, transitions, and Legacy mapping. | | [`core/README.md`](core/README.md) | Shared configuration, request model, and multi-machine deployment settings. | | [`scheduler/README.md`](scheduler/README.md) | Global routing, KDN / Proxy pool management, and Scheduler control plane. | | [`proxy/README.md`](proxy/README.md) | Local Instance pool, capability-aware registration, heartbeat recovery, prepare / ready queues, injection strategy, and Proxy resource APIs. | diff --git a/core/state_models.py b/core/state_models.py new file mode 100644 index 0000000..fa9ad75 --- /dev/null +++ b/core/state_models.py @@ -0,0 +1,298 @@ +"""Versioned, immutable lifecycle contracts for cache and data-plane work.""" + +from __future__ import annotations + +import hashlib +import json +import re +import uuid +from enum import Enum +from typing import ClassVar, Dict, Mapping, Optional, Set, Tuple, Type + +from pydantic import BaseModel, ConfigDict, Field, field_validator + +SCHEMA_VERSION = "1" + + +class ArtifactState(str, Enum): + PENDING = "pending" + BUILDING = "building" + STAGING = "staging" + READY = "ready" + FAILED = "failed" + DELETING = "deleting" + DELETED = "deleted" + + +class ReplicaState(str, Enum): + PENDING = "pending" + STAGING = "staging" + READY = "ready" + FAILED = "failed" + EVICTING = "evicting" + DELETED = "deleted" + + +class ReplicaHealth(str, Enum): + UNKNOWN = "unknown" + HEALTHY = "healthy" + DEGRADED = "degraded" + UNHEALTHY = "unhealthy" + + +class DataPlaneTaskState(str, Enum): + PENDING = "pending" + QUEUED = "queued" + LEASED = "leased" + RUNNING = "running" + SUCCEEDED = "succeeded" + FAILED = "failed" + CANCELLED = "cancelled" + EXPIRED = "expired" + + +class QueueWorkState(str, Enum): + PENDING = "pending" + BLOCKED = "blocked" + READY = "ready" + QUEUED = "queued" + RUNNING = "running" + SUCCEEDED = "succeeded" + FAILED = "failed" + CANCELLED = "cancelled" + SKIPPED = "skipped" + + +LifecycleState = ArtifactState | ReplicaState | DataPlaneTaskState | QueueWorkState + +ARTIFACT_TRANSITIONS = { + ArtifactState.PENDING: {ArtifactState.BUILDING, ArtifactState.FAILED, ArtifactState.DELETING}, + ArtifactState.BUILDING: {ArtifactState.STAGING, ArtifactState.FAILED, ArtifactState.DELETING}, + ArtifactState.STAGING: {ArtifactState.READY, ArtifactState.FAILED, ArtifactState.DELETING}, + ArtifactState.READY: {ArtifactState.BUILDING, ArtifactState.FAILED, ArtifactState.DELETING}, + ArtifactState.FAILED: {ArtifactState.BUILDING, ArtifactState.DELETING}, + ArtifactState.DELETING: {ArtifactState.DELETED, ArtifactState.FAILED}, + ArtifactState.DELETED: set(), +} +REPLICA_TRANSITIONS = { + ReplicaState.PENDING: {ReplicaState.STAGING, ReplicaState.FAILED, ReplicaState.EVICTING}, + ReplicaState.STAGING: {ReplicaState.READY, ReplicaState.FAILED, ReplicaState.EVICTING}, + ReplicaState.READY: {ReplicaState.STAGING, ReplicaState.FAILED, ReplicaState.EVICTING}, + ReplicaState.FAILED: {ReplicaState.STAGING, ReplicaState.EVICTING}, + ReplicaState.EVICTING: {ReplicaState.DELETED, ReplicaState.FAILED}, + ReplicaState.DELETED: set(), +} +TASK_TRANSITIONS = { + DataPlaneTaskState.PENDING: {DataPlaneTaskState.QUEUED, DataPlaneTaskState.CANCELLED}, + DataPlaneTaskState.QUEUED: {DataPlaneTaskState.LEASED, DataPlaneTaskState.RUNNING, DataPlaneTaskState.CANCELLED, DataPlaneTaskState.EXPIRED}, + DataPlaneTaskState.LEASED: {DataPlaneTaskState.RUNNING, DataPlaneTaskState.QUEUED, DataPlaneTaskState.CANCELLED, DataPlaneTaskState.EXPIRED}, + DataPlaneTaskState.RUNNING: {DataPlaneTaskState.SUCCEEDED, DataPlaneTaskState.FAILED, DataPlaneTaskState.CANCELLED}, + DataPlaneTaskState.FAILED: {DataPlaneTaskState.QUEUED, DataPlaneTaskState.CANCELLED}, + DataPlaneTaskState.SUCCEEDED: set(), DataPlaneTaskState.CANCELLED: set(), DataPlaneTaskState.EXPIRED: set(), +} +WORK_TRANSITIONS = { + QueueWorkState.PENDING: {QueueWorkState.BLOCKED, QueueWorkState.READY, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED}, + QueueWorkState.BLOCKED: {QueueWorkState.READY, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED}, + QueueWorkState.READY: {QueueWorkState.QUEUED, QueueWorkState.RUNNING, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED}, + QueueWorkState.QUEUED: {QueueWorkState.RUNNING, QueueWorkState.CANCELLED}, + QueueWorkState.RUNNING: {QueueWorkState.SUCCEEDED, QueueWorkState.FAILED, QueueWorkState.CANCELLED}, + QueueWorkState.FAILED: {QueueWorkState.READY, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED}, + QueueWorkState.SUCCEEDED: set(), QueueWorkState.CANCELLED: set(), QueueWorkState.SKIPPED: set(), +} +TRANSITIONS: Mapping[Type[Enum], Mapping[Enum, Set[Enum]]] = { + ArtifactState: ARTIFACT_TRANSITIONS, ReplicaState: REPLICA_TRANSITIONS, + DataPlaneTaskState: TASK_TRANSITIONS, QueueWorkState: WORK_TRANSITIONS, +} +TERMINAL_STATES = { + ArtifactState.DELETED, ReplicaState.DELETED, + DataPlaneTaskState.SUCCEEDED, DataPlaneTaskState.CANCELLED, DataPlaneTaskState.EXPIRED, + QueueWorkState.SUCCEEDED, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED, +} +RETRYABLE_STATES = { + ArtifactState.FAILED, ReplicaState.FAILED, DataPlaneTaskState.FAILED, QueueWorkState.FAILED, +} + + +def allowed_target_states(state: LifecycleState) -> Tuple[LifecycleState, ...]: + table = TRANSITIONS.get(type(state)) + if table is None: + raise TypeError("state must be a lifecycle state enum") + return tuple(sorted(table[state], key=lambda item: item.value)) + + +def is_terminal_state(state: LifecycleState) -> bool: + return state in TERMINAL_STATES + + +def is_retryable_state(state: LifecycleState) -> bool: + return state in RETRYABLE_STATES + + +class InvalidStateTransitionDetail(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + code: str = "invalid_state_transition" + entity_type: str + entity_id: str + current_state: str + target_state: str + allowed_targets: Tuple[str, ...] + reason: str + terminal: bool + retryable: bool + + +class InvalidStateTransition(ValueError): + def __init__(self, detail: InvalidStateTransitionDetail): + self.detail = detail + super().__init__(detail.model_dump_json()) + + +def validate_state_transition( + current: LifecycleState, target: LifecycleState, *, entity_type: str, entity_id: str, +) -> bool: + allowed = allowed_target_states(current) + if type(target) is type(current) and target == current: + return True + if type(target) is not type(current) or target not in allowed: + target_value = target.value if isinstance(target, Enum) else str(target) + raise InvalidStateTransition(InvalidStateTransitionDetail( + entity_type=entity_type, entity_id=entity_id, current_state=current.value, + target_state=target_value, allowed_targets=tuple(item.value for item in allowed), + reason="terminal_state" if is_terminal_state(current) else "transition_not_allowed", + terminal=is_terminal_state(current), retryable=is_retryable_state(current), + )) + return True + + +class TransitionResult(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + previous_state: str + state: str + changed: bool + + +class _StateModel(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + schema_version: str = SCHEMA_VERSION + _entity_type: ClassVar[str] + _id_field: ClassVar[str] + + @field_validator("schema_version") + @classmethod + def validate_schema_version(cls, value: str) -> str: + if value != SCHEMA_VERSION: + raise ValueError("unsupported schema_version") + return value + + def transition_to(self, target_state: LifecycleState): + current = self.state + validate_state_transition(current, target_state, entity_type=self._entity_type, entity_id=getattr(self, self._id_field)) + if type(target_state) is type(current) and target_state == current: + return self, TransitionResult(previous_state=current.value, state=current.value, changed=False) + return self.model_copy(update={"state": target_state}), TransitionResult( + previous_state=current.value, state=target_state.value, changed=True, + ) + + +def _non_empty(value: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise ValueError("identifier must not be empty") + return value + + +_USER_INFO_PATTERN = re.compile(r"[^/@\s:]+:[^/@\s]+@") +_SECRET_LOCATION_MARKERS = ("://", "password=", "credential=", "token=") + + +def validate_location_key(value: Optional[str]) -> Optional[str]: + """Reject connection strings and obvious secret-bearing Replica identities.""" + if value is None: + return None + if not isinstance(value, str) or not value.strip(): + raise ValueError("location_key must be a non-empty opaque value") + lowered = value.lower() + if any(marker in lowered for marker in _SECRET_LOCATION_MARKERS) or _USER_INFO_PATTERN.search(value): + raise ValueError("location_key must not contain credentials or connection strings") + return value + + +class CacheArtifact(_StateModel): + artifact_id: str + knowledge_id: str + capability_fingerprint: Optional[str] = None + artifact_variant: str = "default" + state: ArtifactState = ArtifactState.PENDING + _entity_type = "cache_artifact" + _id_field = "artifact_id" + _ids = field_validator("artifact_id", "knowledge_id", "artifact_variant")(_non_empty) + + +class CacheReplica(_StateModel): + replica_id: str + artifact_id: str + data_plane_id: Optional[str] = None + backend_type: Optional[str] = None + location_key: Optional[str] = None + state: ReplicaState = ReplicaState.PENDING + health: ReplicaHealth = ReplicaHealth.UNKNOWN + _entity_type = "cache_replica" + _id_field = "replica_id" + _ids = field_validator("replica_id", "artifact_id")(_non_empty) + _location = field_validator("location_key")(validate_location_key) + + +def _task_id() -> str: + return f"dpt:{uuid.uuid4().hex}" + + +def _work_id() -> str: + return f"qwork:{uuid.uuid4().hex}" + + +class DataPlaneTask(_StateModel): + task_id: str = Field(default_factory=_task_id) + operation: Optional[str] = None + artifact_id: Optional[str] = None + source_replica_id: Optional[str] = None + target_replica_id: Optional[str] = None + state: DataPlaneTaskState = DataPlaneTaskState.PENDING + _entity_type = "data_plane_task" + _id_field = "task_id" + _id = field_validator("task_id")(_non_empty) + + +class QueueWork(_StateModel): + work_id: str = Field(default_factory=_work_id) + work_type: Optional[str] = None + resource_class: Optional[str] = None + state: QueueWorkState = QueueWorkState.PENDING + _entity_type = "queue_work" + _id_field = "work_id" + _id = field_validator("work_id")(_non_empty) + + +def _digest(prefix: str, identity: Dict[str, Optional[str]]) -> str: + canonical = json.dumps(identity, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + return f"{prefix}:sha256:{hashlib.sha256(canonical.encode('utf-8')).hexdigest()}" + + +def artifact_id( + knowledge_id: str, capability_fingerprint: Optional[str] = None, + artifact_variant: str = "default", schema_version: str = SCHEMA_VERSION, +) -> str: + normalized = _non_empty(knowledge_id).strip() + return _digest("artifact", {"schema_version": schema_version, "knowledge_id": normalized, + "capability_fingerprint": capability_fingerprint, "artifact_variant": _non_empty(artifact_variant)}) + + +def legacy_artifact_id(knowledge_id: str) -> str: + return artifact_id(knowledge_id, capability_fingerprint=None) + + +def replica_id( + artifact_id_value: str, data_plane_id: Optional[str] = None, + backend_type: Optional[str] = None, location_key: Optional[str] = None, +) -> str: + location_key = validate_location_key(location_key) + return _digest("replica", {"artifact_id": _non_empty(artifact_id_value), + "data_plane_id": data_plane_id, "backend_type": backend_type, "location_key": location_key}) diff --git a/doc/state-models.md b/doc/state-models.md new file mode 100644 index 0000000..1fb4498 --- /dev/null +++ b/doc/state-models.md @@ -0,0 +1,74 @@ +# Cache lifecycle and task state models + +Issue #139 defines immutable, schema-version `"1"` mechanism contracts for CacheArtifact, CacheReplica, DataPlaneTask, and QueueWork. Policy engines may consume these objects, but policy is outside this contract. + +## Wire values and purposes + +| Object | Purpose | Values | +|---|---|---| +| Artifact | capability-specific knowledge materialization | `pending`, `building`, `staging`, `ready`, `failed`, `deleting`, `deleted` | +| Replica | data-plane placement of an Artifact | `pending`, `staging`, `ready`, `failed`, `evicting`, `deleted` | +| Replica health | observation independent of lifecycle | `unknown`, `healthy`, `degraded`, `unhealthy` | +| DataPlaneTask | data creation, movement, or removal work | `pending`, `queued`, `leased`, `running`, `succeeded`, `failed`, `cancelled`, `expired` | +| QueueWork | scheduler-local work | `pending`, `blocked`, `ready`, `queued`, `running`, `succeeded`, `failed`, `cancelled`, `skipped` | + +## Complete transitions + +| Lifecycle | Source | Targets | +|---|---|---| +| Artifact | pending | building, failed, deleting | +| Artifact | building | staging, failed, deleting | +| Artifact | staging | ready, failed, deleting | +| Artifact | ready | building, failed, deleting | +| Artifact | failed | building, deleting | +| Artifact | deleting | deleted, failed | +| Artifact | deleted | none | +| Replica | pending | staging, failed, evicting | +| Replica | staging | ready, failed, evicting | +| Replica | ready | staging, failed, evicting | +| Replica | failed | staging, evicting | +| Replica | evicting | deleted, failed | +| Replica | deleted | none | +| DataPlaneTask | pending | queued, cancelled | +| DataPlaneTask | queued | leased, running, cancelled, expired | +| DataPlaneTask | leased | running, queued, cancelled, expired | +| DataPlaneTask | running | succeeded, failed, cancelled | +| DataPlaneTask | failed | queued, cancelled | +| DataPlaneTask | succeeded, cancelled, expired | none | +| QueueWork | pending | blocked, ready, cancelled, skipped | +| QueueWork | blocked | ready, cancelled, skipped | +| QueueWork | ready | queued, running, cancelled, skipped | +| QueueWork | queued | running, cancelled | +| QueueWork | running | succeeded, failed, cancelled | +| QueueWork | failed | ready, cancelled, skipped | +| QueueWork | succeeded, cancelled, skipped | none | + +Same-state transitions are idempotent. Terminal states are Artifact `deleted`, Replica `deleted`, DataPlaneTask `succeeded`/`cancelled`/`expired`, and QueueWork `succeeded`/`cancelled`/`skipped`. Every `failed` state is retryable and non-terminal. + +Invalid transitions expose typed detail: + +```json +{"code":"invalid_state_transition","entity_type":"cache_artifact","entity_id":"artifact:...","current_state":"deleted","target_state":"pending","allowed_targets":[],"reason":"terminal_state","terminal":true,"retryable":false} +``` + +## Stable IDs + +Artifact IDs have format `artifact:sha256:<64 lowercase hex>` and hash canonical schema version, normalized knowledge ID, nullable capability fingerprint, and Artifact variant. The Legacy helper uses a null capability fingerprint. Replica IDs have format `replica:sha256:<64 lowercase hex>` and hash Artifact ID, nullable data-plane ID, nullable backend type, and nullable opaque non-secret location key. Opaque locations may use values such as `kid` or `node-a/cache-1`; connection strings, embedded user-info, and `password=`, `credential=`, or `token=` values are rejected. Generated IDs are `dpt:<32 lowercase UUID4 hex>` and `qwork:<32 lowercase UUID4 hex>`; restored non-empty IDs are accepted. Credentials, passwords, secret URLs, and raw Redis keys are not identity inputs. + +## Legacy mapping + +Compatibility is always `unknown`: Legacy metadata cannot establish capability compatibility. When checked, only `/` is authoritative; `kv_rel_dir` is validated metadata. + +| kv_ready | Runtime directory | Artifact | Replica | Health | Warning | +|---|---|---|---|---|---| +| false | missing | pending | pending | unknown | none | +| false | exists | staging | staging | unknown | `legacy_files_without_ready_confirmation` | +| true | exists | ready | ready | healthy | none | +| true | missing | ready | failed | unhealthy | `legacy_replica_directory_missing` | +| true | not checked | ready | ready | unknown | none | + +Non-empty `kv_rel_dir` unequal to `kid` adds `legacy_kv_rel_dir_mismatch`. Mapping changes neither SQLite nor files. + +## Boundary and non-goals + +This is lifecycle mechanism, not scheduling policy. Issue #139 does not add issue #140 protocol APIs, a persistent Artifact catalog, ExecutionGraph scheduling, routing/injection changes, database tables, or final operation and resource-class enums. diff --git a/kdn_server/legacy_state.py b/kdn_server/legacy_state.py new file mode 100644 index 0000000..b6f2ae9 --- /dev/null +++ b/kdn_server/legacy_state.py @@ -0,0 +1,120 @@ +"""Read-only projection of Legacy ``kv_ready`` rows onto lifecycle models.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any, Literal, Optional, Tuple + +from pydantic import BaseModel, ConfigDict + +from core.state_models import ( + ArtifactState, CacheArtifact, CacheReplica, ReplicaHealth, ReplicaState, + legacy_artifact_id, replica_id, +) + + +def _row_value(row: Any, key: str, default: Any = None) -> Any: + """Read dictionaries, sqlite3.Row objects, and attribute-backed records safely.""" + getter = getattr(row, "get", None) + if callable(getter): + return getter(key, default) + try: + return row[key] + except (KeyError, IndexError, TypeError): + return getattr(row, key, default) + + +class LegacyStateWarning(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + code: str + message: str + + +class LegacyKVStateView(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + source: Literal["legacy_kv_ready"] = "legacy_kv_ready" + compatibility_status: Literal["unknown"] = "unknown" + artifact: CacheArtifact + replica: CacheReplica + 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 in {".", ".."} or path.is_absolute() or len(path.parts) != 1: + raise ValueError(f"{field} must be a non-empty single path component") + return component + + +def _normalize_kv_ready(value: Any) -> bool: + if value is False or (type(value) is int and value == 0) or (type(value) is str and value == "0"): + return False + if value is True or (type(value) is int and value == 1) or (type(value) is str and value == "1"): + return True + raise ValueError("kv_ready must be one of False, 0, '0', True, 1, or '1'") + + +def map_legacy_kv_state(row: Any, kv_root: Optional[str | Path] = None) -> LegacyKVStateView: + """Map a Legacy row without changing SQLite or the filesystem. + + Passing ``kv_root=None`` explicitly means that no filesystem check was performed. + """ + kid = _safe_component(_row_value(row, "kid"), "kid") + ready = _normalize_kv_ready(_row_value(row, "kv_ready", 0)) + raw_rel_dir = _row_value(row, "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) + + exists: Optional[bool] = None + if kv_root is not None: + 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") + exists = runtime_directory.is_dir() + + warnings = [] + if rel_dir is not None and rel_dir != kid: + warnings.append(LegacyStateWarning( + code="legacy_kv_rel_dir_mismatch", + message="Legacy kv_rel_dir differs from the runtime kid directory.", + )) + + if ready: + artifact_state = ArtifactState.READY + if exists is False: + replica_state, health = ReplicaState.FAILED, ReplicaHealth.UNHEALTHY + warnings.append(LegacyStateWarning( + code="legacy_replica_directory_missing", + message="Legacy kv_ready is set but the runtime kid directory is missing.", + )) + else: + replica_state = ReplicaState.READY + health = ReplicaHealth.HEALTHY if exists is True else ReplicaHealth.UNKNOWN + elif exists is True: + artifact_state, replica_state = ArtifactState.STAGING, ReplicaState.STAGING + health = ReplicaHealth.UNKNOWN + warnings.append(LegacyStateWarning( + code="legacy_files_without_ready_confirmation", + message="Runtime kid directory exists without Legacy kv_ready confirmation.", + )) + else: + artifact_state, replica_state, health = ( + ArtifactState.PENDING, ReplicaState.PENDING, ReplicaHealth.UNKNOWN, + ) + + artifact_identifier = legacy_artifact_id(kid) + artifact = CacheArtifact( + artifact_id=artifact_identifier, knowledge_id=kid, + capability_fingerprint=None, state=artifact_state, + ) + replica = CacheReplica( + replica_id=replica_id(artifact_identifier, location_key=kid), + artifact_id=artifact_identifier, location_key=kid, + state=replica_state, health=health, + ) + return LegacyKVStateView(artifact=artifact, replica=replica, warnings=tuple(warnings)) diff --git a/test/README.md b/test/README.md index efb0fc2..030f88d 100644 --- a/test/README.md +++ b/test/README.md @@ -2,6 +2,13 @@ `test/` contains local entrypoints, smoke scripts, and historical utility tests for CacheRoute development. The main `demo_*.py` entrypoints add the repository root to Python's module search path, so the documented commands can be launched directly from the `test` directory without manually setting `PYTHONPATH`. +Focused CPU-only lifecycle validation from the repository root: + +```bash +pytest -q test/test_state_models.py +pytest -q test/test_legacy_kv_state_mapping.py +``` + Install the complete application and development dependency set from the repository root before running the commands in this document: ```bash diff --git a/test/test_legacy_kv_state_mapping.py b/test/test_legacy_kv_state_mapping.py new file mode 100644 index 0000000..6943624 --- /dev/null +++ b/test/test_legacy_kv_state_mapping.py @@ -0,0 +1,128 @@ +import importlib.util +import sqlite3 +from pathlib import Path + +import pytest +from pydantic import ValidationError + +from core.state_models import ArtifactState, ReplicaHealth, ReplicaState +from kdn_server.legacy_state import map_legacy_kv_state +_spec = importlib.util.spec_from_file_location("proxy_queue_knowledge", Path(__file__).parents[1] / "proxy/queue/knowledge.py") +_knowledge = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(_knowledge) +classify_kdn_items = _knowledge.classify_kdn_items + + +def row(kid="kid", kv_ready=1, kv_rel_dir="kid"): + return {"kid": kid, "kv_ready": kv_ready, "kv_rel_dir": kv_rel_dir} + + +def codes(view): + return {warning.code for warning in view.warnings} + + +@pytest.mark.parametrize("ready,create,artifact,replica,health,warning", [ + (0, False, ArtifactState.PENDING, ReplicaState.PENDING, ReplicaHealth.UNKNOWN, None), + (0, True, ArtifactState.STAGING, ReplicaState.STAGING, ReplicaHealth.UNKNOWN, "legacy_files_without_ready_confirmation"), + (1, True, ArtifactState.READY, ReplicaState.READY, ReplicaHealth.HEALTHY, None), + (1, False, ArtifactState.READY, ReplicaState.FAILED, ReplicaHealth.UNHEALTHY, "legacy_replica_directory_missing"), +]) +def test_complete_filesystem_mapping(tmp_path, ready, create, artifact, replica, health, warning): + if create: + (tmp_path / "kid").mkdir() + view = map_legacy_kv_state(row(kv_ready=ready), tmp_path) + assert (view.artifact.state, view.replica.state, view.replica.health) == (artifact, replica, health) + assert warning is None or warning in codes(view) + + +def test_ready_without_filesystem_check_has_unknown_health(): + view = map_legacy_kv_state(row(), None) + assert view.artifact.state == ArtifactState.READY + assert view.replica.state == ReplicaState.READY + assert view.replica.health == ReplicaHealth.UNKNOWN + + +@pytest.mark.parametrize("value,expected", [(False, False), (0, False), ("0", False), (True, True), (1, True), ("1", True)]) +def test_explicit_kv_ready_normalization(tmp_path, value, expected): + view = map_legacy_kv_state(row(kv_ready=value), tmp_path) + assert (view.artifact.state == ArtifactState.READY) is expected + + +@pytest.mark.parametrize("value", [None, "", "true", 2, -1, [], {}]) +def test_invalid_kv_ready_rejected(tmp_path, value): + with pytest.raises(ValueError, match="kv_ready"): + map_legacy_kv_state(row(kv_ready=value), tmp_path) + + +def test_runtime_kid_is_authoritative_and_mismatch_warns(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.location_key == "kid" + assert "legacy_kv_rel_dir_mismatch" in codes(view) + + +def test_matching_metadata_has_no_mismatch_warning(tmp_path): + (tmp_path / "kid").mkdir() + assert "legacy_kv_rel_dir_mismatch" not in codes(map_legacy_kv_state(row(), tmp_path)) + + +def test_sqlite_row_is_mapped_without_conversion_or_mutation(tmp_path): + (tmp_path / "kid").mkdir() + connection = sqlite3.connect(":memory:") + connection.row_factory = sqlite3.Row + connection.execute("CREATE TABLE legacy (kid TEXT, kv_ready INTEGER, kv_rel_dir TEXT)") + connection.execute("INSERT INTO legacy VALUES (?, ?, ?)", ("kid", 1, "kid")) + sqlite_row = connection.execute("SELECT kid, kv_ready, kv_rel_dir FROM legacy").fetchone() + view = map_legacy_kv_state(sqlite_row, tmp_path) + assert view.artifact.knowledge_id == "kid" + assert view.artifact.state == ArtifactState.READY + assert view.replica.state == ReplicaState.READY + assert view.replica.health == ReplicaHealth.HEALTHY + assert view.replica.location_key == "kid" + assert view.artifact.artifact_id and view.replica.replica_id + assert codes(view) == set() + assert connection.execute("SELECT COUNT(*) FROM legacy").fetchone()[0] == 1 + connection.close() + + +def test_dot_cannot_make_root_healthy(tmp_path): + view = map_legacy_kv_state(row(kv_rel_dir="."), tmp_path) + assert view.replica.state == ReplicaState.FAILED + assert "legacy_kv_rel_dir_mismatch" in codes(view) + + +@pytest.mark.parametrize("field,value", [("kid", "../outside"), ("kv_rel_dir", "../outside")]) +def test_escaped_paths_rejected(tmp_path, field, value): + data = row() + data[field] = value + with pytest.raises(ValueError, match=field): + map_legacy_kv_state(data, tmp_path) + + +def test_ids_are_stable_and_projection_is_deeply_frozen(tmp_path): + first, second = map_legacy_kv_state(row(), tmp_path), map_legacy_kv_state(row(), tmp_path) + assert first.artifact.artifact_id == second.artifact.artifact_id + assert first.replica.replica_id == second.replica.replica_id + assert first.source == "legacy_kv_ready" and first.compatibility_status == "unknown" + with pytest.raises(ValidationError, match="Instance is frozen"): + first.compatibility_status = "compatible" + with pytest.raises(ValidationError, match="Instance is frozen"): + first.replica.state = ReplicaState.READY + + +def test_mapping_does_not_write(tmp_path): + marker = tmp_path / "marker" + marker.write_bytes(b"unchanged") + before = [(path.name, path.stat().st_mtime_ns, path.read_bytes()) for path in tmp_path.iterdir()] + map_legacy_kv_state(row(), tmp_path) + after = [(path.name, path.stat().st_mtime_ns, path.read_bytes()) for path in tmp_path.iterdir()] + assert after == before + + +def test_existing_proxy_classification_is_unchanged(): + items = [{"kid": "ready", "kv_ready": 1}, {"kid": "text", "kv_ready": 0}] + result = classify_kdn_items(["ready", "text", "miss"], items, ["miss"]) + assert [item["kid"] for item in result["kv_ready_items"]] == ["ready"] + assert [item["kid"] for item in result["text_only_items"]] == ["text"] + assert result["miss_ids"] == ["miss"] diff --git a/test/test_state_models.py b/test/test_state_models.py new file mode 100644 index 0000000..fe39395 --- /dev/null +++ b/test/test_state_models.py @@ -0,0 +1,171 @@ +import re + +import pytest +from pydantic import ValidationError + +from core.state_models import ( + ARTIFACT_TRANSITIONS, REPLICA_TRANSITIONS, TASK_TRANSITIONS, WORK_TRANSITIONS, + ArtifactState, CacheArtifact, CacheReplica, DataPlaneTask, DataPlaneTaskState, + InvalidStateTransition, QueueWork, QueueWorkState, ReplicaHealth, ReplicaState, + allowed_target_states, artifact_id, is_retryable_state, is_terminal_state, replica_id, + validate_state_transition, +) + + +ENUM_VALUES = { + ArtifactState: ["pending", "building", "staging", "ready", "failed", "deleting", "deleted"], + ReplicaState: ["pending", "staging", "ready", "failed", "evicting", "deleted"], + ReplicaHealth: ["unknown", "healthy", "degraded", "unhealthy"], + DataPlaneTaskState: ["pending", "queued", "leased", "running", "succeeded", "failed", "cancelled", "expired"], + QueueWorkState: ["pending", "blocked", "ready", "queued", "running", "succeeded", "failed", "cancelled", "skipped"], +} + + +def artifact(state=ArtifactState.PENDING): + return CacheArtifact(artifact_id="artifact:restored", knowledge_id="kid", state=state) + + +def replica(state=ReplicaState.PENDING): + return CacheReplica(replica_id="replica:restored", artifact_id="artifact:restored", state=state) + + +def test_exact_enum_wire_values(): + for enum_type, values in ENUM_VALUES.items(): + assert [item.value for item in enum_type] == values + + +@pytest.mark.parametrize("model", [artifact(), replica(), DataPlaneTask(), QueueWork()]) +def test_json_round_trip(model): + assert type(model).model_validate_json(model.model_dump_json()) == model + + +def test_unknown_enum_and_extra_field_rejected(): + with pytest.raises(ValidationError): + artifact().model_copy(update={"state": "unknown"}).model_validate( + {**artifact().model_dump(), "state": "unknown"}) + with pytest.raises(ValidationError, match="Extra inputs"): + CacheArtifact(artifact_id="a", knowledge_id="k", unexpected=True) + + +@pytest.mark.parametrize("model,target", [ + (artifact(), ArtifactState.READY), (replica(), ReplicaState.READY), + (DataPlaneTask(), DataPlaneTaskState.RUNNING), (QueueWork(), QueueWorkState.RUNNING), +]) +def test_lifecycle_models_are_frozen(model, target): + with pytest.raises(ValidationError, match="Instance is frozen"): + model.state = target + + +@pytest.mark.parametrize("factory,table", [ + (artifact, ARTIFACT_TRANSITIONS), (replica, REPLICA_TRANSITIONS), + (lambda state: DataPlaneTask(task_id="task", state=state), TASK_TRANSITIONS), + (lambda state: QueueWork(work_id="work", state=state), WORK_TRANSITIONS), +]) +def test_every_declared_transition_and_same_state(factory, table): + for source_state, targets in table.items(): + source = factory(source_state) + unchanged, result = source.transition_to(source_state) + assert unchanged is source and result.changed is False + for target in targets: + changed, result = source.transition_to(target) + assert source.state == source_state + assert changed.state == target and result.changed is True + + +def test_invalid_terminal_and_cross_enum_transition_details(): + with pytest.raises(InvalidStateTransition) as terminal: + artifact(ArtifactState.DELETED).transition_to(ArtifactState.PENDING) + assert terminal.value.detail.reason == "terminal_state" + assert terminal.value.detail.terminal is True + assert terminal.value.detail.allowed_targets == () + with pytest.raises(InvalidStateTransition) as cross_enum: + artifact().transition_to(ReplicaState.STAGING) + assert cross_enum.value.detail.reason == "transition_not_allowed" + assert cross_enum.value.detail.target_state == "staging" + + +def test_direct_validator_accepts_same_state_but_rejects_cross_enum_same_wire_value(): + assert validate_state_transition( + ArtifactState.PENDING, ArtifactState.PENDING, + entity_type="cache_artifact", entity_id="artifact:restored", + ) is True + with pytest.raises(InvalidStateTransition): + validate_state_transition( + ArtifactState.PENDING, ReplicaState.PENDING, + entity_type="cache_artifact", entity_id="artifact:restored", + ) + assert [item.value for item in allowed_target_states(ArtifactState.PENDING)] == [ + "building", "deleting", "failed", + ] + + +def test_allowed_targets_are_sorted_and_traits_are_centralized(): + for enum_type, values in ENUM_VALUES.items(): + if enum_type is ReplicaHealth: + continue + for state in enum_type: + assert [item.value for item in allowed_target_states(state)] == sorted(item.value for item in allowed_target_states(state)) + assert is_retryable_state(state) is (state.value == "failed") + assert is_terminal_state(ArtifactState.DELETED) + assert is_terminal_state(ReplicaState.DELETED) + for state in (DataPlaneTaskState.SUCCEEDED, DataPlaneTaskState.CANCELLED, DataPlaneTaskState.EXPIRED, + QueueWorkState.SUCCEEDED, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED): + assert is_terminal_state(state) + assert not is_terminal_state(ArtifactState.FAILED) + + +def test_stable_artifact_and_replica_ids_include_all_identity_inputs(): + first = artifact_id(" kid ", "cap", "variant") + assert first == artifact_id("kid", "cap", "variant") + assert re.fullmatch(r"artifact:sha256:[0-9a-f]{64}", first) + assert first != artifact_id("kid", "other", "variant") + base = replica_id(first, "plane", "disk", "location") + assert re.fullmatch(r"replica:sha256:[0-9a-f]{64}", base) + assert len({base, replica_id(first, "other", "disk", "location"), + replica_id(first, "plane", "other", "location"), + replica_id(first, "plane", "disk", "other"), + replica_id(artifact_id("other"), "plane", "disk", "location")}) == 5 + + +@pytest.mark.parametrize("location", ["kid", "node-a/cache-1", "opaque-location"]) +def test_opaque_replica_locations_are_accepted_and_deterministic(location): + identifier = artifact_id("kid") + assert replica_id(identifier, location_key=location) == replica_id(identifier, location_key=location) + assert CacheReplica(replica_id="restored", artifact_id=identifier, location_key=location).location_key == location + + +@pytest.mark.parametrize("location", [ + "redis://host/key", "password=secret", "credential=value", "token=value", + "user:password@host", "", " ", +]) +def test_secret_bearing_replica_locations_are_rejected(location): + identifier = artifact_id("kid") + with pytest.raises(ValueError, match="location_key"): + replica_id(identifier, location_key=location) + with pytest.raises(ValidationError, match="location_key"): + CacheReplica(replica_id="restored", artifact_id=identifier, location_key=location) + safe_serialized = CacheReplica( + replica_id=replica_id(identifier, location_key="kid"), + artifact_id=identifier, location_key="kid", + ).model_dump_json() + assert all(secret not in safe_serialized for secret in ( + "redis://", "password=", "credential=", "token=", "user:password@host", + )) + + +def test_generated_and_restored_ids_and_empty_rejection(): + tasks = {DataPlaneTask().task_id for _ in range(3)} + works = {QueueWork().work_id for _ in range(3)} + assert len(tasks) == 3 and all(re.fullmatch(r"dpt:[0-9a-f]{32}", item) for item in tasks) + assert len(works) == 3 and all(re.fullmatch(r"qwork:[0-9a-f]{32}", item) for item in works) + assert DataPlaneTask(task_id="restored").task_id == "restored" + assert QueueWork(work_id="restored").work_id == "restored" + for constructor, kwargs in [(DataPlaneTask, {"task_id": ""}), (QueueWork, {"work_id": ""}), + (CacheArtifact, {"artifact_id": "", "knowledge_id": "kid"}), + (CacheReplica, {"replica_id": "", "artifact_id": "a"})]: + with pytest.raises(ValidationError): + constructor(**kwargs) + + +def test_model_defaults_are_independent(): + assert DataPlaneTask().model_dump() is not DataPlaneTask().model_dump()