Add basic version: SemIf decision engine, urgency queue, skill tree, skill loop, decision log, dream pass, CLI

semif_agent/skills.py

@@ -0,0 +1,192 @@
+"""The skill tree, registry, and SemIf-driven navigation.
+
+A skill is a leaf reached by a chain of SemIf choices (category -> skill).
+At every level a "create_skill" branch exists; opencode is the authoring tool
+there (deferred to v2, stubbed as CreateSkill).
+
+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
+
+
+@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:
+    """Sentinel for the 'create a missing skill' branch at a tree level."""
+
+    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, request: Request, tree: dict[str, list[Skill]]) -> Skill | CreateSkill:
+    """Descend the tree one SemIf choice per level."""
+    categories = sorted(tree.keys())
+    create = Option("create_skill", "Create a new skill 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],
+    )
+    top_result = engine.call(top)
+    category = top_result.selected
+    if category == "create_skill":
+        return CreateSkill(category=None)
+    skills = tree[category]
+    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],
+    )
+    leaf_result = engine.call(leaf)
+    pick = leaf_result.selected
+    if pick == "create_skill":
+        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)