From 806cf4dca0dcb9a81bf907d3b77d0ad4cc552a0c Mon Sep 17 00:00:00 2001 From: RickZ <121943721+rickisba@users.noreply.github.com> Date: Sat, 25 Jul 2026 19:47:50 +0800 Subject: [PATCH] feat: add shared lifecycle state models --- README.md | 1 + core/state_models.py | 308 +++++++++++++++++++++++++++ doc/state-models.md | 122 +++++++++++ kdn_server/legacy_state.py | 112 ++++++++++ test/README.md | 26 +++ test/test_legacy_kv_state_mapping.py | 65 ++++++ test/test_state_models.py | 147 +++++++++++++ 7 files changed, 781 insertions(+) create mode 100644 core/state_models.py create mode 100644 doc/state-models.md create mode 100644 kdn_server/legacy_state.py create mode 100644 test/test_legacy_kv_state_mapping.py create mode 100644 test/test_state_models.py diff --git a/README.md b/README.md index 8c384fd..f0e83b7 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) | Shared CacheArtifact, CacheReplica, DataPlaneTask, and QueueWork lifecycle contracts and Legacy KV metadata 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..5c3e8e9 --- /dev/null +++ b/core/state_models.py @@ -0,0 +1,308 @@ +"""Shared, policy-neutral lifecycle contracts for cache and execution objects.""" +from __future__ import annotations + +import hashlib +import json +import uuid +from enum import Enum +from typing import Any, Dict, List, Mapping, Optional, Set, Tuple, Type, TypeVar + +from pydantic import BaseModel, ConfigDict, Field, field_validator + + +class _WireEnum(str, Enum): + pass + + +class ArtifactState(_WireEnum): + PENDING = "pending" + BUILDING = "building" + STAGING = "staging" + READY = "ready" + FAILED = "failed" + DELETING = "deleting" + DELETED = "deleted" + + +class ReplicaState(_WireEnum): + PENDING = "pending" + STAGING = "staging" + READY = "ready" + FAILED = "failed" + EVICTING = "evicting" + DELETED = "deleted" + + +class ReplicaHealth(_WireEnum): + UNKNOWN = "unknown" + HEALTHY = "healthy" + DEGRADED = "degraded" + UNHEALTHY = "unhealthy" + + +class DataPlaneTaskState(_WireEnum): + PENDING = "pending" + QUEUED = "queued" + LEASED = "leased" + RUNNING = "running" + SUCCEEDED = "succeeded" + FAILED = "failed" + CANCELLED = "cancelled" + EXPIRED = "expired" + + +class QueueWorkState(_WireEnum): + PENDING = "pending" + BLOCKED = "blocked" + READY = "ready" + QUEUED = "queued" + RUNNING = "running" + SUCCEEDED = "succeeded" + FAILED = "failed" + CANCELLED = "cancelled" + SKIPPED = "skipped" + + +_TRANSITIONS: Dict[Type[_WireEnum], Mapping[_WireEnum, Set[_WireEnum]]] = { + ArtifactState: { + 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(), + }, + ReplicaState: { + 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(), + }, + DataPlaneTaskState: { + 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(), + }, + QueueWorkState: { + 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(), + }, +} + +_TERMINAL = { + ArtifactState: {ArtifactState.DELETED}, ReplicaState: {ReplicaState.DELETED}, + DataPlaneTaskState: {DataPlaneTaskState.SUCCEEDED, DataPlaneTaskState.CANCELLED, DataPlaneTaskState.EXPIRED}, + QueueWorkState: {QueueWorkState.SUCCEEDED, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED}, +} +_RETRYABLE = { + ArtifactState: {ArtifactState.FAILED}, ReplicaState: {ReplicaState.FAILED}, + DataPlaneTaskState: {DataPlaneTaskState.FAILED}, QueueWorkState: {QueueWorkState.FAILED}, +} + + +class StateTransitionDetail(BaseModel): + model_config = ConfigDict(extra="forbid") + code: str = "invalid_state_transition" + entity_type: str + entity_id: str + current_state: str + target_state: str + allowed_targets: List[str] = Field(default_factory=list) + reason: str + terminal: bool + retryable: bool + + +class StateTransitionResult(BaseModel): + model_config = ConfigDict(extra="forbid") + changed: bool + current_state: str + target_state: str + + +class InvalidStateTransition(Exception): + def __init__(self, detail: StateTransitionDetail): + self.detail = detail + super().__init__(detail.model_dump_json()) + + +S = TypeVar("S", bound=_WireEnum) + + +def allowed_target_states(state: S) -> Tuple[S, ...]: + return tuple(sorted(_TRANSITIONS[type(state)][state], key=lambda item: item.value)) # type: ignore[return-value] + + +def is_terminal_state(state: _WireEnum) -> bool: + return state in _TERMINAL[type(state)] + + +def is_retryable_state(state: _WireEnum) -> bool: + return state in _RETRYABLE[type(state)] + + +def validate_state_transition(entity_type: str, entity_id: str, current: S, target: S) -> StateTransitionResult: + if type(current) is not type(target): + raise TypeError("current and target states must use the same state enum") + if current == target: + return StateTransitionResult(changed=False, current_state=current.value, target_state=target.value) + if target not in _TRANSITIONS[type(current)][current]: + raise InvalidStateTransition(StateTransitionDetail( + entity_type=entity_type, entity_id=entity_id, current_state=current.value, + target_state=target.value, allowed_targets=[item.value for item in allowed_target_states(current)], + reason="terminal_state" if is_terminal_state(current) else "transition_not_allowed", + terminal=is_terminal_state(current), retryable=is_retryable_state(current), + )) + return StateTransitionResult(changed=True, current_state=current.value, target_state=target.value) + + +def _canonical_id(prefix: str, values: Dict[str, Any]) -> str: + payload = json.dumps(values, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + return f"{prefix}:sha256:{hashlib.sha256(payload.encode('utf-8')).hexdigest()}" + + +def _required_text(value: str, name: str) -> str: + normalized = value.strip() + if not normalized: + raise ValueError(f"{name} must not be empty") + return normalized + + +def _non_secret_location(value: str) -> str: + normalized = _required_text(value, "location_key") + lowered = normalized.lower() + if "://" in normalized or "password=" in lowered or "credential=" in lowered: + raise ValueError("location_key must be an opaque, non-secret identity") + return normalized + + +def generate_artifact_id(knowledge_id: str, capability_fingerprint: Optional[str], artifact_variant: str = "default", schema_version: str = "1") -> str: + return _canonical_id("artifact", { + "artifact_variant": _required_text(artifact_variant, "artifact_variant"), + "capability_fingerprint": capability_fingerprint.strip() if capability_fingerprint else None, + "knowledge_id": _required_text(knowledge_id, "knowledge_id").lower(), + "schema_version": _required_text(schema_version, "schema_version"), + }) + + +def generate_legacy_artifact_id(knowledge_id: str, artifact_variant: str = "default", schema_version: str = "1") -> str: + """Generate an ID while explicitly retaining unknown compatibility.""" + return generate_artifact_id(knowledge_id, None, artifact_variant, schema_version) + + +def generate_replica_id(artifact_id: str, data_plane_id: Optional[str], backend_type: Optional[str], location_key: Optional[str]) -> str: + return _canonical_id("replica", { + "artifact_id": _required_text(artifact_id, "artifact_id"), + "backend_type": backend_type.strip() if backend_type else None, + "data_plane_id": data_plane_id.strip() if data_plane_id else None, + "location_key": _non_secret_location(location_key) if location_key else None, + }) + + +def generate_data_plane_task_id() -> str: + return f"dpt:{uuid.uuid4().hex}" + + +def generate_queue_work_id() -> str: + return f"qwork:{uuid.uuid4().hex}" + + +class _StateModel(BaseModel): + model_config = ConfigDict(extra="forbid") + + @field_validator("schema_version", check_fields=False) + @classmethod + def _nonempty_schema(cls, value: str) -> str: + return _required_text(value, "schema_version") + + def _transition(self, entity_type: str, entity_id: str, target: _WireEnum): + result = validate_state_transition(entity_type, entity_id, self.state, target) + return self.model_copy(update={"state": target}), result + + +class CacheArtifact(_StateModel): + schema_version: str = "1" + artifact_id: str + knowledge_id: str + capability_fingerprint: Optional[str] = None + artifact_variant: str = "default" + state: ArtifactState = ArtifactState.PENDING + + @field_validator("artifact_id", "knowledge_id", "artifact_variant") + @classmethod + def _nonempty(cls, value: str) -> str: + return _required_text(value, "identifier") + + def transition_to(self, target: ArtifactState): + return self._transition("artifact", self.artifact_id, target) + + +class CacheReplica(_StateModel): + schema_version: str = "1" + 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 + + @field_validator("replica_id", "artifact_id") + @classmethod + def _nonempty(cls, value: str) -> str: + return _required_text(value, "identifier") + + @field_validator("location_key") + @classmethod + def _opaque_location(cls, value: Optional[str]) -> Optional[str]: + if value is None: + return None + return _non_secret_location(value) + + def transition_to(self, target: ReplicaState): + return self._transition("replica", self.replica_id, target) + + +class DataPlaneTask(_StateModel): + schema_version: str = "1" + task_id: str = Field(default_factory=generate_data_plane_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 + + @field_validator("task_id") + @classmethod + def _nonempty(cls, value: str) -> str: + return _required_text(value, "task_id") + + def transition_to(self, target: DataPlaneTaskState): + return self._transition("data_plane_task", self.task_id, target) + + +class QueueWork(_StateModel): + schema_version: str = "1" + work_id: str = Field(default_factory=generate_queue_work_id) + work_type: Optional[str] = None + resource_class: Optional[str] = None + state: QueueWorkState = QueueWorkState.PENDING + + @field_validator("work_id") + @classmethod + def _nonempty(cls, value: str) -> str: + return _required_text(value, "work_id") + + def transition_to(self, target: QueueWorkState): + return self._transition("queue_work", self.work_id, target) diff --git a/doc/state-models.md b/doc/state-models.md new file mode 100644 index 0000000..f3c0ab3 --- /dev/null +++ b/doc/state-models.md @@ -0,0 +1,122 @@ +# Cache and Execution State Models + +This document defines the v0.1.10 shared, serialized mechanism for cache lifecycle and execution state. It does not activate a catalog, data-plane protocol, task registry, execution graph, routing change, or scheduling/maintenance policy. + +## Objects and IDs + +| Object | Purpose | ID rule | +|---|---|---| +| `CacheArtifact` | A model/runtime-specific reusable cache artifact for one knowledge object and variant. | `artifact:sha256:<64 hex>` over compact, key-sorted JSON containing schema version, normalized knowledge ID, capability fingerprint, and artifact variant. The Legacy helper records a null fingerprint; it never invents compatibility data. | +| `CacheReplica` | One physical/logical placement of an Artifact, with lifecycle and health kept separate. | `replica:sha256:<64 hex>` over Artifact ID, data-plane ID, backend type, and an opaque non-secret location key. Credentials and backend-internal Redis keys must not be location identities. | +| `DataPlaneTask` | A future data-plane execution request without a finalized operation enum. | `dpt:<32 lowercase UUID4 hex>`. A restored non-empty external ID is also accepted. | +| `QueueWork` | A future schedulable work item without dependency or priority policy. | `qwork:<32 lowercase UUID4 hex>`. A restored non-empty external ID is also accepted. | + +Canonical JSON uses sorted keys, compact separators, UTF-8, and schema version `"1"`. Changing any identity input changes its deterministic hash. All models reject extra fields. + +## Artifact lifecycle + +Wire values are `pending`, `building`, `staging`, `ready`, `failed`, `deleting`, and `deleted`. + +| Current | Allowed targets | +|---|---| +| pending | building, deleting, failed | +| building | deleting, failed, staging | +| staging | deleting, failed, ready | +| ready | building, deleting, failed | +| failed | building, deleting | +| deleting | deleted, failed | +| deleted | none | + +`deleted` is terminal. `failed` is retryable. `ready` is not terminal because refresh, invalidation, and deletion remain possible. + +## Replica lifecycle and health + +Replica wire values are `pending`, `staging`, `ready`, `failed`, `evicting`, and `deleted`. + +| Current | Allowed targets | +|---|---| +| pending | evicting, failed, staging | +| staging | evicting, failed, ready | +| ready | evicting, failed, staging | +| failed | evicting, staging | +| evicting | deleted, failed | +| deleted | none | + +`deleted` is terminal and `failed` is retryable. Health has the independent wire values `unknown`, `healthy`, `degraded`, and `unhealthy`. For example, a ready Replica can be degraded, and an Artifact can remain ready when one known Replica fails. These values do not prescribe maintenance action. + +## Data-plane task lifecycle + +Wire values are `pending`, `queued`, `leased`, `running`, `succeeded`, `failed`, `cancelled`, and `expired`. + +| Current | Allowed targets | +|---|---| +| pending | cancelled, queued | +| queued | cancelled, expired, leased, running | +| leased | cancelled, expired, queued, running | +| running | cancelled, failed, succeeded | +| failed | cancelled, queued | +| succeeded | none | +| cancelled | none | +| expired | none | + +`succeeded`, `cancelled`, and `expired` are terminal. `failed` is retryable and non-terminal. Retry attempt and lease policies are intentionally undefined. + +## Queue work lifecycle + +Wire values are `pending`, `blocked`, `ready`, `queued`, `running`, `succeeded`, `failed`, `cancelled`, and `skipped`. + +`pending` means dependency evaluation is incomplete; `blocked` waits for dependencies; `ready` is schedulable; `queued` is admitted to a concrete resource; and `running` has begun execution. + +| Current | Allowed targets | +|---|---| +| pending | blocked, cancelled, ready, skipped | +| blocked | cancelled, ready, skipped | +| ready | cancelled, queued, running, skipped | +| queued | cancelled, running | +| running | cancelled, failed, succeeded | +| failed | cancelled, ready, skipped | +| succeeded | none | +| cancelled | none | +| skipped | none | + +`succeeded`, `cancelled`, and `skipped` are terminal. `failed` is retryable and non-terminal. Dependency graphs, resource admission, ordering, and retry policy remain separate from this mechanism. + +## Transition behavior and errors + +Every same-state transition succeeds idempotently with `changed: false`. Other allowed transitions return a new model copy and `changed: true`; the original model is not mutated. Allowed targets are always sorted in machine-readable output. + +An invalid transition raises `InvalidStateTransition` with a typed detail such as: + +```json +{ + "code": "invalid_state_transition", + "entity_type": "artifact", + "entity_id": "artifact:sha256:...", + "current_state": "ready", + "target_state": "pending", + "allowed_targets": ["building", "deleting", "failed"], + "reason": "transition_not_allowed", + "terminal": false, + "retryable": false +} +``` + +A transition away from a terminal state uses `reason: "terminal_state"`. + +## Legacy `kv_ready` adapter + +The adapter is a read-only projection. It does not migrate SQLite, alter `mark_kv_ready`, or change `KV_database/`. + +| `kv_ready` | Directory check | 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 performed | ready | ready | unknown | none | + +Boolean values and integer/string `0` and `1` are normalized explicitly; other values fail. The filesystem wrapper resolves the location below the configured KV root and rejects path escape. Legacy compatibility is always `unknown`, because historical rows lack the complete Instance capability fingerprint. + +## Explicit non-goals + +Issue #139 does not implement persistent Artifact/Replica tables, a KDN task registry, data-plane endpoints or submission protocol, leases, idempotency, a production worker, ExecutionGraph scheduling, queue priorities, retry/eviction policy, trace schema, placement, or routing. Existing Proxy queues and `kv_ready` classification remain active and unchanged. Issue #140 can build a versioned protocol and operation categories on this state contract; policy work must continue to treat lifecycle state as mechanism rather than a decision rule. diff --git a/kdn_server/legacy_state.py b/kdn_server/legacy_state.py new file mode 100644 index 0000000..8801c27 --- /dev/null +++ b/kdn_server/legacy_state.py @@ -0,0 +1,112 @@ +"""Read-only compatibility views for legacy ``kv_ready`` metadata.""" +from __future__ import annotations + +from pathlib import Path +from typing import List, Literal, Optional, Union + +from pydantic import BaseModel, ConfigDict, Field + +from core.state_models import ( + ArtifactState, CacheArtifact, CacheReplica, ReplicaHealth, ReplicaState, + generate_legacy_artifact_id, generate_replica_id, +) + + +class LegacyStateWarning(BaseModel): + model_config = ConfigDict(extra="forbid") + code: str + message: str + + +class LegacyKVStateView(BaseModel): + model_config = ConfigDict(extra="forbid") + source: Literal["legacy_kv_ready"] = "legacy_kv_ready" + compatibility_status: Literal["unknown"] = "unknown" + artifact: CacheArtifact + replica: CacheReplica + warnings: List[LegacyStateWarning] = Field(default_factory=list) + + +def normalize_legacy_kv_ready(value: Union[bool, int, str]) -> bool: + """Normalize SQLite-compatible legacy values without Python truthiness.""" + if isinstance(value, bool): + return value + if isinstance(value, int) and value in (0, 1): + return value == 1 + if isinstance(value, str) and value.strip() in ("0", "1"): + return value.strip() == "1" + raise ValueError("kv_ready must be bool, integer 0/1, or string '0'/'1'") + + +def map_legacy_kv_state( + *, kid: str, kv_ready: Union[bool, int, str], kv_rel_dir: Optional[str], + kv_dumped_keys: Optional[int], kv_updated_at: Optional[int], + replica_directory_exists: Optional[bool], +) -> LegacyKVStateView: + """Purely derive lifecycle state; legacy timing/count fields remain unchanged.""" + del kv_dumped_keys, kv_updated_at # Inputs are accepted but do not imply lifecycle state. + normalized_kid = (kid or "").strip().lower() + if not normalized_kid: + raise ValueError("kid must not be empty") + ready = normalize_legacy_kv_ready(kv_ready) + location_key = (kv_rel_dir or normalized_kid).strip() + if not location_key: + raise ValueError("kv_rel_dir must not be empty when provided") + + warnings: List[LegacyStateWarning] = [] + if ready: + artifact_state = ArtifactState.READY + if replica_directory_exists is False: + replica_state, health = ReplicaState.FAILED, ReplicaHealth.UNHEALTHY + warnings.append(LegacyStateWarning( + code="legacy_replica_directory_missing", + message="Legacy metadata marks KV ready, but the replica directory is missing.", + )) + elif replica_directory_exists is True: + replica_state, health = ReplicaState.READY, ReplicaHealth.HEALTHY + else: + replica_state, health = ReplicaState.READY, ReplicaHealth.UNKNOWN + elif replica_directory_exists is True: + artifact_state, replica_state, health = ArtifactState.STAGING, ReplicaState.STAGING, ReplicaHealth.UNKNOWN + warnings.append(LegacyStateWarning( + code="legacy_files_without_ready_confirmation", + message="Legacy replica files exist without a KV-ready confirmation.", + )) + else: + artifact_state, replica_state, health = ArtifactState.PENDING, ReplicaState.PENDING, ReplicaHealth.UNKNOWN + + artifact_id = generate_legacy_artifact_id(normalized_kid) + replica_id = generate_replica_id(artifact_id, "legacy_kdn", "legacy_file", location_key) + return LegacyKVStateView( + artifact=CacheArtifact( + artifact_id=artifact_id, knowledge_id=normalized_kid, + capability_fingerprint=None, state=artifact_state, + ), + replica=CacheReplica( + replica_id=replica_id, artifact_id=artifact_id, data_plane_id="legacy_kdn", + backend_type="legacy_file", location_key=location_key, + state=replica_state, health=health, + ), + warnings=warnings, + ) + + +def map_legacy_kv_state_from_filesystem( + *, kv_root: Union[str, Path], kid: str, kv_ready: Union[bool, int, str], + kv_rel_dir: Optional[str], kv_dumped_keys: Optional[int], kv_updated_at: Optional[int], +) -> LegacyKVStateView: + """Resolve the legacy directory below ``kv_root`` and reject path escapes.""" + root = Path(kv_root).resolve() + relative = kv_rel_dir or (kid or "").strip().lower() + if not relative: + raise ValueError("legacy replica location must not be empty") + candidate = (root / relative).resolve() + try: + candidate.relative_to(root) + except ValueError as exc: + raise ValueError("legacy replica location escapes the configured KV root") from exc + return map_legacy_kv_state( + kid=kid, kv_ready=kv_ready, kv_rel_dir=relative, + kv_dumped_keys=kv_dumped_keys, kv_updated_at=kv_updated_at, + replica_directory_exists=candidate.is_dir(), + ) diff --git a/test/README.md b/test/README.md index efb0fc2..ff32ceb 100644 --- a/test/README.md +++ b/test/README.md @@ -65,6 +65,32 @@ The default single-machine demo ports are: | `Prefill_calculation.py` | Prefill-time calculation helper. | | `quick_start_docker.sh` | Convenience script for container-oriented startup. | +## Focused state-model validation + +Run the CPU-only lifecycle contract and Legacy `kv_ready` adapter tests from the repository root: + +```bash +python3 -m compileall core kdn_server proxy +pytest -q test/test_state_models.py +pytest -q test/test_legacy_kv_state_mapping.py +pytest -q \ + test/test_state_models.py \ + test/test_legacy_kv_state_mapping.py \ + test/test_instance_capability.py \ + test/test_instance_capability_registration.py \ + proxy/resource/test_instance_pool_gpu.py +``` + +These checks cover exact wire values, deterministic IDs, every allowed transition, structured transition failures, terminal/retryable traits, safe Legacy filesystem mapping, unknown Legacy compatibility, and unchanged Proxy KDN classification. Success is a zero exit status with all listed tests passing. Any validation error, ID instability, unexpected transition acceptance, path-escape acceptance, or changed classification indicates failure. + +For the largest practical CPU-only suite, use: + +```bash +pytest -q --ignore=test/test_kv_injector_reuse.py +``` + +The ignored historical script contacts an external vLLM-compatible endpoint during collection. + ## Focused Instance capability validation The Instance capability contract covers deterministic fingerprints, structured compatibility results, capability-aware registration, and heartbeat recovery while preserving legacy payloads. diff --git a/test/test_legacy_kv_state_mapping.py b/test/test_legacy_kv_state_mapping.py new file mode 100644 index 0000000..b1ad128 --- /dev/null +++ b/test/test_legacy_kv_state_mapping.py @@ -0,0 +1,65 @@ +import pytest + +from core.state_models import ArtifactState, ReplicaHealth, ReplicaState +from kdn_server.legacy_state import map_legacy_kv_state, map_legacy_kv_state_from_filesystem +from proxy.queue.knowledge import classify_kdn_items + + +@pytest.mark.parametrize("value", [False, 0, "0"]) +def test_not_ready_without_directory(value): + view = map_legacy_kv_state(kid="KID", kv_ready=value, kv_rel_dir="KID", kv_dumped_keys=None, kv_updated_at=None, replica_directory_exists=False) + assert (view.artifact.state, view.replica.state, view.replica.health) == (ArtifactState.PENDING, ReplicaState.PENDING, ReplicaHealth.UNKNOWN) + assert view.compatibility_status == "unknown" and view.artifact.capability_fingerprint is None + + +def test_not_ready_with_files_warns(): + view = map_legacy_kv_state(kid="kid", kv_ready=False, kv_rel_dir="kid", kv_dumped_keys=2, kv_updated_at=1, replica_directory_exists=True) + assert (view.artifact.state, view.replica.state) == (ArtifactState.STAGING, ReplicaState.STAGING) + assert view.warnings[0].code == "legacy_files_without_ready_confirmation" + + +@pytest.mark.parametrize("value", [True, 1, "1"]) +def test_ready_with_directory(value): + view = map_legacy_kv_state(kid="kid", kv_ready=value, kv_rel_dir="kid", kv_dumped_keys=2, kv_updated_at=1, replica_directory_exists=True) + assert (view.artifact.state, view.replica.state, view.replica.health) == (ArtifactState.READY, ReplicaState.READY, ReplicaHealth.HEALTHY) + + +def test_ready_with_missing_directory_keeps_artifact_ready(): + view = map_legacy_kv_state(kid="kid", kv_ready=True, kv_rel_dir="kid", kv_dumped_keys=2, kv_updated_at=1, replica_directory_exists=False) + assert view.artifact.state == ArtifactState.READY + assert (view.replica.state, view.replica.health) == (ReplicaState.FAILED, ReplicaHealth.UNHEALTHY) + assert view.warnings[0].code == "legacy_replica_directory_missing" + + +def test_ready_without_filesystem_check_has_unknown_health(): + view = map_legacy_kv_state(kid="kid", kv_ready=True, kv_rel_dir="kid", kv_dumped_keys=None, kv_updated_at=None, replica_directory_exists=None) + assert (view.replica.state, view.replica.health) == (ReplicaState.READY, ReplicaHealth.UNKNOWN) + + +@pytest.mark.parametrize("value", [2, -1, "false", "true", "", None, object()]) +def test_invalid_ready_values_fail(value): + with pytest.raises(ValueError, match="kv_ready"): + map_legacy_kv_state(kid="kid", kv_ready=value, kv_rel_dir="kid", kv_dumped_keys=None, kv_updated_at=None, replica_directory_exists=None) + + +def test_filesystem_wrapper_and_escape_rejection(tmp_path): + (tmp_path / "kid").mkdir() + view = map_legacy_kv_state_from_filesystem(kv_root=tmp_path, kid="kid", kv_ready=1, kv_rel_dir="kid", kv_dumped_keys=1, kv_updated_at=1) + assert view.replica.health == ReplicaHealth.HEALTHY + with pytest.raises(ValueError, match="escapes"): + map_legacy_kv_state_from_filesystem(kv_root=tmp_path, kid="kid", kv_ready=1, kv_rel_dir="../outside", kv_dumped_keys=1, kv_updated_at=1) + + +def test_mapping_ids_are_stable_and_mapper_does_not_write(tmp_path): + kwargs = dict(kv_root=tmp_path, kid="kid", kv_ready=0, kv_rel_dir="kid", kv_dumped_keys=None, kv_updated_at=None) + first, second = map_legacy_kv_state_from_filesystem(**kwargs), map_legacy_kv_state_from_filesystem(**kwargs) + assert first.artifact.artifact_id == second.artifact.artifact_id + assert first.replica.replica_id == second.replica.replica_id + assert list(tmp_path.iterdir()) == [] + + +def test_existing_proxy_classification_is_unchanged(): + result = classify_kdn_items(["ready", "text", "missing"], [{"id": "ready", "kv_ready": 1}, {"id": "text", "kv_ready": 0}], ["missing"]) + assert [item["id"] for item in result["kv_ready_items"]] == ["ready"] + assert [item["id"] for item in result["text_only_items"]] == ["text"] + assert result["miss_ids"] == ["missing"] diff --git a/test/test_state_models.py b/test/test_state_models.py new file mode 100644 index 0000000..369475b --- /dev/null +++ b/test/test_state_models.py @@ -0,0 +1,147 @@ +import re + +import pytest +from pydantic import ValidationError + +from core.state_models import ( + ArtifactState, CacheArtifact, CacheReplica, DataPlaneTask, DataPlaneTaskState, + InvalidStateTransition, QueueWork, QueueWorkState, ReplicaHealth, ReplicaState, + StateTransitionDetail, allowed_target_states, generate_artifact_id, + generate_data_plane_task_id, generate_queue_work_id, generate_replica_id, + is_retryable_state, is_terminal_state, +) + + +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"]), +] + + +@pytest.mark.parametrize("enum_type,values", ENUM_VALUES) +def test_exact_enum_values(enum_type, values): + assert [item.value for item in enum_type] == values + with pytest.raises(ValueError): + enum_type("future_value") + + +def _models(): + aid = generate_artifact_id("kid", "sha256:cap") + rid = generate_replica_id(aid, "plane", "file", "kid") + return [ + CacheArtifact(artifact_id=aid, knowledge_id="kid"), + CacheReplica(replica_id=rid, artifact_id=aid), DataPlaneTask(), QueueWork(), + ] + + +@pytest.mark.parametrize("index", range(4)) +def test_json_round_trip_and_forbid_extra(index): + model = _models()[index] + assert type(model).model_validate_json(model.model_dump_json()) == model + with pytest.raises(ValidationError): + type(model).model_validate({**model.model_dump(), "unexpected": True}) + + +def test_unknown_state_is_rejected_and_mutable_defaults_are_independent(): + with pytest.raises(ValidationError): + DataPlaneTask(state="not_a_state") + left = StateTransitionDetail(entity_type="x", entity_id="1", current_state="a", target_state="b", reason="x", terminal=False, retryable=False) + right = StateTransitionDetail(entity_type="x", entity_id="2", current_state="a", target_state="b", reason="x", terminal=False, retryable=False) + left.allowed_targets.append("c") + assert right.allowed_targets == [] + + +def test_stable_artifact_ids_and_canonical_inputs(): + first = generate_artifact_id(" KID ", "sha256:cap", "variant") + assert first == generate_artifact_id("kid", "sha256:cap", "variant") + assert re.fullmatch(r"artifact:sha256:[0-9a-f]{64}", first) + assert first != generate_artifact_id("other", "sha256:cap", "variant") + assert first != generate_artifact_id("kid", "sha256:other", "variant") + assert first != generate_artifact_id("kid", "sha256:cap", "other") + + +def test_replica_ids_are_stable_and_sensitive_without_exposing_inputs(): + args = ("artifact:sha256:" + "a" * 64, "plane", "file", "opaque-location") + replica_id = generate_replica_id(*args) + assert replica_id == generate_replica_id(*args) + assert re.fullmatch(r"replica:sha256:[0-9a-f]{64}", replica_id) + for changed in [ + (args[0], "other-plane", args[2], args[3]), (args[0], args[1], "redis", args[3]), + (args[0], args[1], args[2], "other-location"), + ]: + assert replica_id != generate_replica_id(*changed) + secret = "redis://user:password@example/key" + with pytest.raises(ValueError, match="opaque, non-secret"): + generate_replica_id(args[0], "plane", "redis", secret) + with pytest.raises(ValidationError, match="opaque, non-secret"): + CacheReplica(replica_id=replica_id, artifact_id=args[0], location_key=secret) + + +def test_generated_and_restored_execution_ids(): + task_ids = {generate_data_plane_task_id() for _ in range(10)} + work_ids = {generate_queue_work_id() for _ in range(10)} + assert len(task_ids) == len(work_ids) == 10 + assert all(re.fullmatch(r"dpt:[0-9a-f]{32}", item) for item in task_ids) + assert all(re.fullmatch(r"qwork:[0-9a-f]{32}", item) for item in work_ids) + assert DataPlaneTask(task_id="restored-task").task_id == "restored-task" + assert QueueWork(work_id="restored-work").work_id == "restored-work" + with pytest.raises(ValidationError): DataPlaneTask(task_id=" ") + with pytest.raises(ValidationError): QueueWork(work_id="") + + +TRANSITIONS = [ + (CacheArtifact, ArtifactState, {"pending": "building failed deleting", "building": "staging failed deleting", "staging": "ready failed deleting", "ready": "building failed deleting", "failed": "building deleting", "deleting": "deleted failed"}), + (CacheReplica, ReplicaState, {"pending": "staging failed evicting", "staging": "ready failed evicting", "ready": "staging failed evicting", "failed": "staging evicting", "evicting": "deleted failed"}), + (DataPlaneTask, DataPlaneTaskState, {"pending": "queued cancelled", "queued": "leased running cancelled expired", "leased": "running queued cancelled expired", "running": "succeeded failed cancelled", "failed": "queued cancelled"}), + (QueueWork, QueueWorkState, {"pending": "blocked ready cancelled skipped", "blocked": "ready cancelled skipped", "ready": "queued running cancelled skipped", "queued": "running cancelled", "running": "succeeded failed cancelled", "failed": "ready cancelled skipped"}), +] + + +def _make(model_type, state): + aid = generate_artifact_id("kid", None) + if model_type is CacheArtifact: return model_type(artifact_id=aid, knowledge_id="kid", state=state) + if model_type is CacheReplica: return model_type(replica_id="replica", artifact_id=aid, state=state) + return model_type(state=state) + + +@pytest.mark.parametrize("model_type,enum_type,mapping", TRANSITIONS) +def test_every_allowed_transition_and_same_state(model_type, enum_type, mapping): + for source, targets in mapping.items(): + model = _make(model_type, enum_type(source)) + for target in targets.split(): + changed, result = model.transition_to(enum_type(target)) + assert changed.state == enum_type(target) and result.changed + assert model.state == enum_type(source) + same, result = model.transition_to(enum_type(source)) + assert same.state == model.state and not result.changed + assert [item.value for item in allowed_target_states(enum_type(source))] == sorted(targets.split()) + + +@pytest.mark.parametrize("model_type,enum_type,mapping", TRANSITIONS) +def test_invalid_and_terminal_transition_details(model_type, enum_type, mapping): + source = next(iter(mapping)) + invalid = next(item for item in enum_type if item.value != source and item.value not in mapping[source].split()) + with pytest.raises(InvalidStateTransition) as caught: + _make(model_type, enum_type(source)).transition_to(invalid) + assert caught.value.detail.code == "invalid_state_transition" + assert caught.value.detail.reason == "transition_not_allowed" + assert caught.value.detail.allowed_targets == sorted(mapping[source].split()) + terminal = next(item for item in enum_type if is_terminal_state(item)) + target = next(item for item in enum_type if item != terminal) + with pytest.raises(InvalidStateTransition) as terminal_error: + _make(model_type, terminal).transition_to(target) + assert terminal_error.value.detail.reason == "terminal_state" + assert terminal_error.value.detail.terminal is True + + +def test_terminal_and_retryable_traits(): + expected_terminal = {ArtifactState.DELETED, ReplicaState.DELETED, DataPlaneTaskState.SUCCEEDED, DataPlaneTaskState.CANCELLED, DataPlaneTaskState.EXPIRED, QueueWorkState.SUCCEEDED, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED} + expected_retryable = {ArtifactState.FAILED, ReplicaState.FAILED, DataPlaneTaskState.FAILED, QueueWorkState.FAILED} + for enum_type in (ArtifactState, ReplicaState, DataPlaneTaskState, QueueWorkState): + for state in enum_type: + assert is_terminal_state(state) == (state in expected_terminal) + assert is_retryable_state(state) == (state in expected_retryable) + assert all(not is_terminal_state(state) for state in expected_retryable)