50 lines
1.8 KiB
Python
50 lines
1.8 KiB
Python
"""Server-owned business instructions, selected from the persisted task workflow."""
|
|
|
|
from hashlib import sha256
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
|
|
SKILL_ROOT = Path(__file__).resolve().parents[2] / "resources" / "work-order-skills"
|
|
WORKFLOW_SKILLS = {
|
|
"scada_anomaly": "scada-anomaly",
|
|
"burst_report": "burst-event-handling",
|
|
"water_quality_complaint": "water-quality-complaint",
|
|
"general": "appointment-follow-up",
|
|
}
|
|
|
|
|
|
def skill_name_for_task(workflow: str | None, replay_context: dict | None = None) -> str | None:
|
|
if workflow == "burst_report" and replay_context:
|
|
return "scada-anomaly"
|
|
return WORKFLOW_SKILLS.get(workflow)
|
|
|
|
|
|
def read_work_order_skill(
|
|
workflow: str | None, *, include_content: bool = False, replay_context: dict | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Read only packaged, allowlisted instructions; never interpret a task path."""
|
|
name = skill_name_for_task(workflow, replay_context)
|
|
if name is None:
|
|
return {"available": False, "workflow": workflow, "reason": "No business skill is configured for this workflow"}
|
|
content = (SKILL_ROOT / name / "SKILL.md").read_text(encoding="utf-8")
|
|
_, frontmatter, body = content.split("---", 2)
|
|
header = yaml.safe_load(frontmatter)
|
|
if header["name"] != name:
|
|
raise ValueError("Packaged work-order skill name does not match its directory")
|
|
result = {
|
|
"available": True,
|
|
"kind": "business_instructions",
|
|
"workflow": workflow,
|
|
"name": header["name"],
|
|
"description": header["description"],
|
|
"version": str(header["metadata"]["version"]),
|
|
"content_sha256": sha256(content.encode("utf-8")).hexdigest(),
|
|
"read": {"section": "skill"},
|
|
}
|
|
if include_content:
|
|
result["instructions"] = body.strip()
|
|
return result
|