"""The skill tree, registry, and SemIf-driven navigation. A skill is a leaf reached by a chain of SemIf choices (category -> skill). The category level carries a "create_category" branch and the leaf level a "create_skill" branch; both are stubs that log a suggestion event to the trace (deferred to v2 — no actual authoring yet). Only the real skills live here; navigation uses the real decision engine. """ from __future__ import annotations import json import os from dataclasses import dataclass, field from pathlib import Path from typing import Callable from .decisions import DecisionRequest, Option, Request from .engine import SemIfEngine from .log import DecisionLog from .trace import TraceLog @dataclass class ActionResult: action_log: str new_state: str @dataclass class Prediction: """The predict phase: a forecast plus any SemIf decisions it made.""" text: str decisions: list[tuple[DecisionRequest, object]] = field(default_factory=list) @dataclass class ActionContext: engine: SemIfEngine config: dict @dataclass class Skill: name: str category: str description: str cost_budget: float = 1.0 predict: Callable[[ActionContext, Request], Prediction] = field( default=lambda ctx, req: Prediction(text="") ) act: Callable[[ActionContext, Request, Prediction], ActionResult] = field( default=lambda ctx, req, pred: ActionResult("", "") ) @dataclass class CreateSkill: """Stub for a missing-category/skill suggestion at a tree level. Navigation logs the suggestion event to the trace; actual authoring is deferred to v2. `category` is None for a new-category suggestion, else the category that needs the new skill. """ category: str | None = None def compose_state(request: Request, current: str | None = None) -> str: parts = [request.text] if current: parts.append(f"[current process: {current}]") return " ".join(parts) def _contacts(ctx: ActionContext) -> list[dict]: path = Path(ctx.config.get("contacts", "data/contacts.json")) if not path.is_file(): return [] return json.loads(path.read_text()) def _email_predict(ctx: ActionContext, request: Request) -> Prediction: contacts = _contacts(ctx) if not contacts: return Prediction(text="no contacts available", decisions=[]) decision = DecisionRequest( state=compose_state(request), question="Which contact is the intended recipient?", options=[Option(c["name"], c.get("description", "")) for c in contacts] + [Option("none", "None of the listed contacts.")], ) result = ctx.engine.call(decision) return Prediction(text=f"recipient is {result.selected}", decisions=[(decision, result)]) def _email_compose(ctx: ActionContext, request: Request, prediction: Prediction) -> ActionResult: recipient = prediction.text.removeprefix("recipient is ") if recipient == "no contacts available" or recipient == "none": return ActionResult( action_log="email.compose aborted: recipient not resolved.", new_state=request.text, ) drafts = Path(ctx.config.get("drafts", "data/drafts")) drafts.mkdir(parents=True, exist_ok=True) target = drafts / f"{request.id}.txt" target.write_text(f"To: {recipient}\nBody: {request.text}\n") return ActionResult( action_log=f"email.compose: wrote draft {target} for {recipient!r}.", new_state=f"Draft written to {target.name} for {recipient}.", ) def _response_reject(ctx: ActionContext, request: Request, prediction: Prediction) -> ActionResult: message = f"Rejected: I cannot act on this while busy ({request.text})." return ActionResult(action_log=f"response.reject: {message}", new_state=message) def _tracking_check(ctx: ActionContext, request: Request, prediction: Prediction) -> ActionResult: path = Path(ctx.config.get("packages", "data/packages.json")) if not path.is_file(): return ActionResult( action_log="tracking.check aborted: no packages file.", new_state=request.text, ) packages = json.loads(path.read_text()) lines = [f"{p.get('id')}: {p.get('status')}" for p in packages] report = "Tracking statuses:\n" + "\n".join(lines) return ActionResult(action_log="tracking.check: " + report, new_state=report) def build_skills(config: dict) -> list[Skill]: skills = config.get("skills", {}) return [ Skill( name="email.compose", category="email", description="Compose and dispatch an email.", predict=_email_predict, act=_email_compose, cost_budget=float(skills.get("email", {}).get("cost_budget", 1.0)), ), Skill( name="response.reject", category="response", description="Politely reject a request because the agent is busy.", act=_response_reject, ), Skill( name="tracking.check", category="tracking", description="Check the delivery status of a package.", act=_tracking_check, ), ] def build_tree(skills: list[Skill]) -> dict[str, list[Skill]]: tree: dict[str, list[Skill]] = {} for skill in skills: tree.setdefault(skill.category, []).append(skill) for category in tree: tree[category].sort(key=lambda s: s.name) return tree def navigate( engine: SemIfEngine, log: DecisionLog, trace: TraceLog, request: Request, tree: dict[str, list[Skill]], ) -> Skill | CreateSkill: """Descend the tree one SemIf choice per level. Every choice is logged. The category level offers a "create_category" branch and the leaf level a "create_skill" branch; both log a suggestion event to the trace and return a CreateSkill stub (actual authoring is deferred to v2). """ categories = sorted(tree.keys()) create_category = Option("create_category", "Suggest a new category for this.") top = DecisionRequest( state=compose_state(request), question="Which top-level category handles this request?", options=[Option(c, c) for c in categories] + [create_category], ) top_result = engine.call(top) log.append(top, top_result, extra={"phase": "navigate:category", "run_id": request.id}) category = top_result.selected if category == "create_category": trace.append( "create_category", request.id, state=top.state, question=top.question, options=[o.id for o in top.options], selected=top_result.selected, probs=top_result.probs, ) return CreateSkill(category=None) skills = tree[category] create_skill = Option("create_skill", "Suggest creating a new skill.") leaf = DecisionRequest( state=compose_state(request, current=category), question=f"Within {category}, which skill?", options=[Option(s.name, s.description) for s in skills] + [create_skill], ) leaf_result = engine.call(leaf) log.append(leaf, leaf_result, extra={"phase": "navigate:leaf", "run_id": request.id}) pick = leaf_result.selected if pick == "create_skill": trace.append( "skill_needed", request.id, category=category, state=leaf.state, question=leaf.question, options=[o.id for o in leaf.options], selected=leaf_result.selected, probs=leaf_result.probs, ) return CreateSkill(category=category) return next(s for s in skills if s.name == pick) def tree_summary(tree: dict[str, list[Skill]]) -> str: lines = [] for category in sorted(tree): names = ", ".join(s.name for s in tree[category]) lines.append(f" {category}: {names}") return "\n".join(lines)