File size: 3,796 Bytes
cb505ff | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | """Documentation-driven acceptance tests for roadmap phases 6 through 12."""
from __future__ import annotations
import inspect
import pytest
from src.core.graph import _worker_skill_node
from src.core.policy import PlatformGateService, PolicyEngine
from src.runtime.cdm_runtime import TrendAnalysis
from src.runtime.execution_context import ExecutionContext
from src.runtime.metrics import RuntimeMetrics
from src.runtime.worker_runtime import WorkerRuntime
from src.skills.adapters import LEGACY_WORKER_FACTORIES
from src.skills.registry import default_skill_registry
from src.skills.runtime import SkillRuntime
def test_phase6_registry_contains_all_non_clinical_worker_skills():
registry = default_skill_registry()
assert {
"caregiver-support",
"diabetes-research",
"diabetes-dietary-coach",
} <= {skill.id for skill in registry.list_active()}
def test_phase7_clinical_manifests_enforce_role_and_intent():
registry = default_skill_registry()
runtime = SkillRuntime(registry)
runtime.select("clinical-diagnosis-support", role="clinician", intent="diagnosis")
with pytest.raises(PermissionError):
runtime.select("clinical-diagnosis-support", role="patient", intent="diagnosis")
with pytest.raises(PermissionError):
runtime.select("clinical-diagnosis-support", role="clinician", intent="treatment")
def test_phase8_cdm_trend_is_typed_and_survives_worker_flush():
trend = TrendAnalysis(summary="Glucose is stable.", metrics={"Glucose": {"trend": "stable"}})
context = ExecutionContext(user_query="Review my trends")
context.metadata["cdm_trend_analysis"] = trend.as_dict()
worker = WorkerRuntime(default_skill_registry())
worker.load_skill("cdm-health-coach")
worker.flush_and_destroy()
assert context.metadata["cdm_trend_analysis"] == {
"trend_analysis": "Glucose is stable.",
"trend_data": {"Glucose": {"trend": "stable"}},
}
def test_phase9_failed_gate_cannot_publish_artifact():
state = {"gates": {"emergency": {"decision": "passed"}, "validation": {"decision": "failed"}, "safety": {"decision": "passed"}}}
assert PlatformGateService.can_publish(state) is False
invalidated = PlatformGateService.invalidate_artifact(
{"response_artifacts": [{"status": "generated"}]}, "validation"
)
assert invalidated["response_artifacts"][0]["status"] == "invalidated_by_validation"
def test_phase10_graph_worker_surface_is_skill_based():
source = inspect.getsource(_worker_skill_node)
assert "WorkerRuntime" in source
assert "create_worker_adapter" in source
def test_phase11_has_no_legacy_worker_singletons_in_graph_runtime():
import src.core.graph as graph
assert "agent_instances" not in inspect.getsource(graph)
def test_phase12_runtime_metrics_capture_lifecycle_and_context_counters():
metrics = RuntimeMetrics()
metrics.record("runtime.created", runtime="WorkerRuntime")
metrics.record("skill.loaded", runtime="WorkerRuntime", duration_ms=1.2)
metrics.record("skill.executed", prompt_tokens=12, context_tokens=8)
metrics.record("runtime.destroyed", runtime="WorkerRuntime")
snapshot = metrics.snapshot()
assert snapshot["event_count"] == 4
assert snapshot["durations_ms"]["skill.loaded"] == 1.2
assert snapshot["active_runtime_slots"] == 0
def test_policy_emergency_detection_is_deterministic():
assert PolicyEngine.check_emergency("The patient is unconscious") is True
assert PolicyEngine.check_emergency("The patient wants meal planning") is False
def test_phase6_compatibility_factories_are_explicitly_scoped():
assert set(LEGACY_WORKER_FACTORIES) >= {
"caregiver-support",
"diabetes-research",
"diabetes-dietary-coach",
} |