From 4e0e0534c40a3075cf7994ec0868bdc3dd352e04 Mon Sep 17 00:00:00 2001 From: semif-agent Date: Wed, 23 Sep 2026 03:01:45 -0500 Subject: [PATCH] Add basic version: SemIf decision engine, urgency queue, skill tree, skill loop, decision log, dream pass, CLI --- .gitignore | 5 + IDEA.md | 115 ++++++++++++++++ config.json | 21 +++ data/contacts.json | 5 + data/packages.json | 5 + pyproject.toml | 16 +++ semif_agent/__init__.py | 3 + semif_agent/cli.py | 158 +++++++++++++++++++++ semif_agent/decisions.py | 85 ++++++++++++ semif_agent/dream.py | 123 +++++++++++++++++ semif_agent/engine.py | 93 +++++++++++++ semif_agent/llm.py | 86 ++++++++++++ semif_agent/log.py | 78 +++++++++++ semif_agent/queue.py | 83 +++++++++++ semif_agent/scheduler.py | 212 +++++++++++++++++++++++++++++ semif_agent/skill.py | 65 +++++++++ semif_agent/skills.py | 192 ++++++++++++++++++++++++++ tests/integration/test_pipeline.py | 79 +++++++++++ tests/test_contract.py | 42 ++++++ tests/test_dream.py | 89 ++++++++++++ tests/test_queue.py | 74 ++++++++++ 21 files changed, 1629 insertions(+) create mode 100644 .gitignore create mode 100644 IDEA.md create mode 100644 config.json create mode 100644 data/contacts.json create mode 100644 data/packages.json create mode 100644 pyproject.toml create mode 100644 semif_agent/__init__.py create mode 100644 semif_agent/cli.py create mode 100644 semif_agent/decisions.py create mode 100644 semif_agent/dream.py create mode 100644 semif_agent/engine.py create mode 100644 semif_agent/llm.py create mode 100644 semif_agent/log.py create mode 100644 semif_agent/queue.py create mode 100644 semif_agent/scheduler.py create mode 100644 semif_agent/skill.py create mode 100644 semif_agent/skills.py create mode 100644 tests/integration/test_pipeline.py create mode 100644 tests/test_contract.py create mode 100644 tests/test_dream.py create mode 100644 tests/test_queue.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..d5a355e --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +__pycache__/ +*.pyc +.venv/ +data/decisions.jsonl +data/drafts/ \ No newline at end of file diff --git a/IDEA.md b/IDEA.md new file mode 100644 index 0000000..fffc0dd --- /dev/null +++ b/IDEA.md @@ -0,0 +1,115 @@ +# Semif Agent + +A local desktop CLI agent whose control flow is a single decision model (SemIf). Inputs are gated, scored, queued, and dispatched through a skill tree; every decision is logged as a labeled training row, and the decision model is fine-tuned at regular intervals on the accumulated prediction-vs-observation cost. + +## Engine: SemIf (local decision layer) + +- **SemIf** (formerly OpenJev, TheoLeeCJ, MIT) is a local rebuild of Jev's interface pattern. One forward pass reads typed option logits directly from a model; no answer sentence, no decoding loop. +- Contract: input `{state, question, options[]}`, output per-option probabilities conditional on the supplied options. Probabilities are conditional on the option set, not absolute confidence — calibrate per workload (SemIf ships per-workload temperature scaling). +- **Why local**: desktop target, data stays in-network (SemIf's data boundary), runs on a home GPU (or CPU via llama.cpp). Jev (hosted) is not in scope. +- Decision model is pinned to an exact revision; every swap is auditable via prompt hashes in SemIf's output. + +## Core loop + +``` +intake (typed / events / timers / other skills) + └─ contains_request? (gate: is this even actionable?) + └─ choice (SemIf: "should this interrupt the current process?") + ├─ yes → preempt current, requeue it with state preserved + └─ no → score (SemIf: "how urgent?") + └─ priority queue, sorted by urgency desc +``` + +- **Scheduler**: run head of queue when idle. A new interrupt displaces the current process, which itself requeues preserving its state. +- **Concurrency** (TODO resolved): single execution slot for v1. SemIf's shared-state mode (one prefetched state branched across many criteria in parallel) is the lever for parallelizing decisions later, not concurrent process execution. +- **Tie-breaking**: FIFO within equal urgency, recency as secondary key. +- **Priority ageing**: queued items decay upward over time so they can't starve. +- **Bounds**: max queue depth; overflow rejects with a notification. + +## Decision templates + +All decisions are SemIf calls: `{state, question, options[]}`. State is the current process + input. Options are typed with descriptions. + +- **`choice`** — binary: `interrupt` / `defer`. Interrupt iff `P(interrupt) >= τ`. +- **`score`** — ordinal urgency: `critical` / `high` / `medium` / `low`, mapped to numeric weights for sorting. +- **skill navigation** — at each tree level: choose category / descend / `create_skill`. +- **`read_next()`** — argument selection within a skill (e.g., which contact is "girlfriend"). + +**LLM/SemIf boundary**: SemIf for fast, repeated, low-latency decisions (gating, scoring, routing, argument selection). LLM for generation and assessment (email body, self-assessment summary). Never the reverse. + +## Skill tree + +- Structure: categories → skills → actions. Top level listed at each level. +- Navigation is a chain of SemIf choices, one per level, descending until a leaf skill matches. +- At each level a `create_skill` branch exists: **opencode authors the skill** (its only role) and drops a skill manifest into the registry. The new skill becomes a leaf immediately. + +### Skill manifest +- name, category, description, allowed inputs, action list, cost budget, decision log reference. + +## Skill anatomy + +Every skill run follows the same loop: + +1. **observe baseline** — capture state relevant to the skill. +2. **make prediction** — predict outcome; log it. +3. **act** — execute actions (possibly calling `read_next()`-style SemIf decisions for arguments). +4. **observe outcome** — capture post-action state. +5. **self-assess** (LLM) — "have I succeeded?" → success/failure + summary + **updated request** back to the priority queue. + +- The self-assessment result **labels** every SemIf decision made during that run (`choice`, `score`, navigation, argument selection): the predicted distribution vs. the observed outcome is one labeled training row. +- Per-skill cost budget: a run that exceeds it fails fast and requeues with a degraded goal. + +## Learning / fine-tuning pipeline + +- **Log**: every decision row → `{state, question, options, predicted_probs, observed_outcome}` — SemIf `decisions.jsonl`-compatible, so logs replay directly into the scorer. +- **Cost function**: loss between the predicted distribution and the observed outcome — cross-entropy / NLL under a proper scoring rule, plus optional calibration loss. This is the training signal. +- **Labels**: self-assessment output by default; human confirmations/corrections override and become high-weight labels. +- **"Dreaming"** — a cronjob at a regular interval: accumulate logged rows → compute cost over the period → fine-tune the decision model on them → CI/CD validates (accuracy / ECE on a held-out slice, prompt-hash regression) → swap in the new pinned model revision. +- **GPU offload**: training runs on a beefier GPU; the running agent keeps a frozen inference revision until a swap validates. +- Cold start: no labels yet — seed from self-assessed runs only; human overrides accelerate early calibration. + +## Worked examples + +### Example 1 +- current_process: none +- input: "send my girlfriend an email that says that I'm going to be late to the party" +- contains_request: yes +- interrupt_current: yes +- skill_selection: email => compose => transcribe(input): [code block that uses SemIf `read_next()` to determine "girlfriend" from contacts, plus a normally functioning LLM for the email body] + +### Example 2 +- current_process: driving car +- input: "send my girlfriend an email that says that I'm going to be late to the party" +- interrupt_current: no +- priority: .21 +- skill_selection: response => rejection(input) + +### Example 3 +- current_process: driving car +- input: "holy shit stop!" +- contains_request: yes +- interrupt_current: yes +- skill_selection: driving => decelerate(input) + +### Example 4 (skill creation) +- current_process: none +- input: "tell me if my package was delivered" +- contains_request: yes +- interrupt_current: yes +- skill_selection: tracking => (no leaf) => create_skill +- opencode authors `tracking.check_delivery` from the input + registry conventions → manifest registered → requeue → skill_selection: tracking => check_delivery(input) + +### Example 5 (self-assessment + dream) +- current_process: email compose +- input: "send my girlfriend an email…" → email compose succeeds but sends to the wrong contact (`read_next` picked wrong) +- self-assess: fail → labels the `read_next` decision row (predicted_probs vs wrong outcome) +- cost computed, logged +- next dream cycle: row included in fine-tune → model revision swapped → `read_next` recalibrated + +## Open questions + +- **Safety / authority**: which inputs may interrupt high-stakes processes (driving-grade)? Is interrupt a per-skill permission, not a global default? +- **Queue persistence**: is the priority queue in-memory or durable across restarts? +- **Input taxonomy**: enumerate intake sources and their `contains_request` gating semantics. +- **Privacy boundary**: all data stays local — confirm no telemetry even for dreaming. +- **Override UX**: how do human confirmations/corrections get surfaced and captured cheaply? \ No newline at end of file diff --git a/config.json b/config.json new file mode 100644 index 0000000..42e8186 --- /dev/null +++ b/config.json @@ -0,0 +1,21 @@ +{ + "tau": 0.6, + "max_reentries": 3, + "queue": {"max_size": 100, "age_rate": 0.01}, + "engine": { + "backend": "llamacpp", + "source": "Qwen/Qwen3.5-4B", + "revision": "851bf6e806efd8d0a36b00ddf55e13ccb7b8cd0a", + "gguf": "/mnt/models/Qwen3.5-4B-Q4_K_M.gguf", + "context_tokens": 4096, + "threads": null + }, + "llm": {"base_url": "http://localhost:11434/v1", "model": "qwen2.5:3b"}, + "skills": { + "email": {"cost_budget": 1.0}, + "contacts": "data/contacts.json", + "drafts": "data/drafts", + "packages": "data/packages.json" + }, + "log": "data/decisions.jsonl" +} \ No newline at end of file diff --git a/data/contacts.json b/data/contacts.json new file mode 100644 index 0000000..fa40c96 --- /dev/null +++ b/data/contacts.json @@ -0,0 +1,5 @@ +[ + {"name": "alice", "description": "Partner; usually home by six."}, + {"name": "bob", "description": "Work colleague on the ops team."}, + {"name": "carol", "description": "Landlord; prefers email."} +] \ No newline at end of file diff --git a/data/packages.json b/data/packages.json new file mode 100644 index 0000000..49efe47 --- /dev/null +++ b/data/packages.json @@ -0,0 +1,5 @@ +[ + {"id": "1Z999AA10123456784", "status": "out for delivery"}, + {"id": "1Z999BB10234567891", "status": "delivered"}, + {"id": "1Z999CC10345678908", "status": "at sorting facility"} +] \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..6adc9b2 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,16 @@ +[build-system] +requires = ["setuptools>=61"] +build-backend = "setuptools.build_meta" + +[project] +name = "semif-agent" +version = "0.1.0" +description = "A local desktop agent whose control flow is one decision model (SemIf)." +requires-python = ">=3.10" +dependencies = [] + +[project.optional-dependencies] +test = ["pytest"] + +[tool.setuptools.packages.find] +include = ["semif_agent*"] \ No newline at end of file diff --git a/semif_agent/__init__.py b/semif_agent/__init__.py new file mode 100644 index 0000000..f54ffc0 --- /dev/null +++ b/semif_agent/__init__.py @@ -0,0 +1,3 @@ +"""Semif agent: a local desktop agent whose control flow is one decision model (SemIf).""" + +__version__ = "0.1.0" diff --git a/semif_agent/cli.py b/semif_agent/cli.py new file mode 100644 index 0000000..16c107b --- /dev/null +++ b/semif_agent/cli.py @@ -0,0 +1,158 @@ +"""CLI entrypoint: interactive REPL, scripted JSONL mode, and subcommands. + +Run on the box with SemIf + a GGUF + a local OpenAI-compatible server: + + python -m semif_agent.cli run # REPL + python -m semif_agent.cli run --script inputs.jsonl + python -m semif_agent.cli dream # prediction-observation cost report + python -m semif_agent.cli skills + python -m semif_agent.cli status + python -m semif_agent.cli relabel +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +from .dream import dream as run_dream +from .engine import EngineConfig, EngineUnavailable, SemIfEngine +from .llm import LLMClient +from .log import DecisionLog +from .scheduler import Scheduler +from .skills import build_skills, build_tree, tree_summary + + +def load_config(path: str = "config.json") -> dict: + return json.loads(Path(path).read_text()) + + +def build_scheduler(config: dict) -> tuple[Scheduler, dict]: + engine = SemIfEngine( + EngineConfig( + backend=config.get("engine", {}).get("backend", "llamacpp"), + source=config.get("engine", {}).get("source", ""), + revision=config.get("engine", {}).get("revision", ""), + gguf=config.get("engine", {}).get("gguf", ""), + context_tokens=int(config.get("engine", {}).get("context_tokens", 4096)), + threads=config.get("engine", {}).get("threads"), + ) + ) + llm = LLMClient( + base_url=config.get("llm", {}).get("base_url", "http://localhost:11434/v1"), + model=config.get("llm", {}).get("model", "qwen2.5:3b"), + ) + log = DecisionLog(config.get("log", "data/decisions.jsonl")) + scheduler = Scheduler( + engine=engine, + llm=llm, + log=log, + config=config, + tau=float(config.get("tau", 0.6)), + max_reentries=int(config.get("max_reentries", 3)), + ) + return scheduler, config + + +def try_warm(scheduler: Scheduler) -> str: + try: + scheduler.engine._ensure_loaded() + return "decision engine loaded." + except EngineUnavailable as exc: + return f"decision engine unavailable: {exc}" + + +def repl(scheduler: Scheduler, config: dict) -> None: + print(try_warm(scheduler)) + print("type a request, or one of: busy | idle | status | skills | dream | relabel | quit") + while True: + try: + line = input("> ").strip() + except (EOFError, KeyboardInterrupt): + print() + break + if not line: + continue + lower = line.lower() + if lower in ("quit", "exit"): + break + if lower == "status": + print(scheduler.status()) + continue + if lower == "skills": + print(tree_summary(build_tree(build_skills(config)))) + continue + if lower == "dream": + print(run_dream(scheduler.log).render()) + continue + if lower.startswith("relabel "): + parts = line.split() + if len(parts) != 3: + print("usage: relabel ") + continue + ok = scheduler.log.relabel(parts[1], parts[2]) + print("relabeled." if ok else f"no row with id {parts[1]}") + continue + if lower == "idle": + scheduler.idle() + print("current process cleared.") + continue + if lower.startswith("busy "): + scheduler.busy(line[5:].strip()) + print("current process set (busy).") + continue + status, detail = scheduler.submit(line) + print(f"[{status}] {detail}") + + +def scripted(scheduler: Scheduler, path: str) -> None: + print(try_warm(scheduler)) + rows = [json.loads(line) for line in Path(path).read_text().splitlines() if line.strip()] + for row in rows: + status, detail = scheduler.submit(str(row["text"]), source=row.get("source", "scripted")) + print(f"[{status}] {detail}") + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="semif-agent") + parser.add_argument("--config", default="config.json") + sub = parser.add_subparsers(dest="command") + + run_p = sub.add_parser("run", help="interactive REPL or scripted input") + run_p.add_argument("--script", default=None, help="JSONL file of {\"text\": ...} rows") + + sub.add_parser("dream", help="compute the prediction-observation cost report") + sub.add_parser("skills", help="list the skill tree") + sub.add_parser("status", help="show current process and queue") + + relabel_p = sub.add_parser("relabel", help="human override of a decision label") + relabel_p.add_argument("id") + relabel_p.add_argument("outcome") + + args = parser.parse_args(argv) + config = load_config(args.config) + scheduler, config = build_scheduler(config) + + if args.command == "run": + if args.script: + scripted(scheduler, args.script) + else: + repl(scheduler, config) + elif args.command == "dream": + print(run_dream(scheduler.log).render()) + elif args.command == "skills": + print(tree_summary(build_tree(build_skills(config)))) + elif args.command == "status": + print(scheduler.status()) + elif args.command == "relabel": + ok = scheduler.log.relabel(args.id, args.outcome) + print("relabeled." if ok else f"no row with id {args.id}") + else: + parser.print_help() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file diff --git a/semif_agent/decisions.py b/semif_agent/decisions.py new file mode 100644 index 0000000..5850f77 --- /dev/null +++ b/semif_agent/decisions.py @@ -0,0 +1,85 @@ +"""The SemIf decision contract shared across the agent. + +A decision is a typed question over a state with declared options; SemIf returns +probabilities conditional on exactly the supplied options. These are not +calibrated confidence values, so callers treat them as conditional scores. +""" + +from __future__ import annotations + +import time +import uuid +from dataclasses import dataclass, field +from typing import Any + + +@dataclass +class Option: + id: str + description: str + + +@dataclass +class DecisionRequest: + """One SemIf decision: state + question + typed options.""" + + state: str + question: str + options: list[Option] + id: str = field(default_factory=lambda: uuid.uuid4().hex[:12]) + + def to_semif_row(self) -> dict: + return { + "id": self.id, + "state": self.state, + "question": self.question, + "options": [{"id": o.id, "description": o.description} for o in self.options], + } + + +@dataclass +class DecisionResult: + """The outcome of one SemIf call: probabilities aligned to option ids.""" + + request: DecisionRequest + option_ids: list[str] + probabilities: list[float] + extra: dict[str, Any] = field(default_factory=dict) + + @property + def probs(self) -> dict[str, float]: + return dict(zip(self.option_ids, self.probabilities)) + + def prob(self, option_id: str) -> float: + index = self.option_ids.index(option_id) + return self.probabilities[index] + + @property + def selected(self) -> str: + return max(self.probs, key=self.probs.get) + + +@dataclass +class Request: + """An incoming input to the agent, before it is gated/scored.""" + + text: str + id: str = field(default_factory=lambda: uuid.uuid4().hex[:12]) + source: str = "typed" + meta: dict[str, Any] = field(default_factory=dict) + received_at: float = field(default_factory=time.time) + priority: float = 0.5 + reentries: int = 0 + resume: dict[str, Any] = field(default_factory=dict) + + def copy_for_requeue(self) -> "Request": + return Request( + text=self.text, + id=self.id, + source=self.source, + meta=dict(self.meta), + received_at=self.received_at, + priority=self.priority, + reentries=self.reentries + 1, + resume=dict(self.resume), + ) diff --git a/semif_agent/dream.py b/semif_agent/dream.py new file mode 100644 index 0000000..598d3a3 --- /dev/null +++ b/semif_agent/dream.py @@ -0,0 +1,123 @@ +"""The dream pass: compute the prediction-observation cost over the decision log. + +Replays `decisions.jsonl` and reports the training signal for the fine-tuning +step: per-row negative log likelihood (NLL) of the observed outcome under the +predicted distribution, overall cross-entropy, accuracy, and binned expected +calibration error (ECE). Human-override labels are weighted higher, matching +the design (self-assessment + human overrides). This computes the signal; the +actual fine-tuning and CI/CD model swap are a separate v2 step. +""" + +from __future__ import annotations + +import math +from dataclasses import dataclass, field + +from .log import DecisionLog + +HUMAN_WEIGHT = 3.0 +SELF_WEIGHT = 1.0 + + +@dataclass +class RowCost: + decision_id: str + predicted: float + observed: str + selected: str + nll: float + weight: float + correct: bool + + +@dataclass +class DreamReport: + rows: list[RowCost] = field(default_factory=list) + skipped: int = 0 + human_overrides: int = 0 + + @property + def cross_entropy(self) -> float | None: + if not self.rows: + return None + total = sum(r.weight for r in self.rows) + return sum(r.weight * r.nll for r in self.rows) / total + + @property + def accuracy(self) -> float | None: + if not self.rows: + return None + return sum(r.correct for r in self.rows) / len(self.rows) + + @property + def ece(self) -> float | None: + """Binned expected calibration error over confidence of the selected option.""" + if not self.rows: + return None + bins: dict[int, list[RowCost]] = {index: [] for index in range(10)} + for row in self.rows: + bin_index = min(int(row.predicted * 10), 9) + bins[bin_index].append(row) + total_weight = sum(r.weight for r in self.rows) + if total_weight <= 0: + return None + error = 0.0 + for index, members in bins.items(): + if not members: + continue + weight = sum(r.weight for r in members) + confidence = sum(r.predicted * r.weight for r in members) / weight + accuracy = sum(r.correct * r.weight for r in members) / weight + error += abs(confidence - accuracy) * weight / total_weight + return error + + def render(self) -> str: + lines = [ + f"decision rows: {len(self.rows)} (skipped {self.skipped})", + f"human overrides: {self.human_overrides}", + ] + ce = self.cross_entropy + acc = self.accuracy + ece = self.ece + lines.append(f"cross-entropy (weighted): {ce:.4f}" if ce is not None else "cross-entropy: n/a") + lines.append(f"accuracy (selected==observed): {acc:.3f}" if acc is not None else "accuracy: n/a") + lines.append(f"ECE (10 bins): {ece:.4f}" if ece is not None else "ECE: n/a") + if self.rows: + worst = sorted(self.rows, key=lambda r: r.nll, reverse=True)[:5] + lines.append("highest-cost rows:") + for row in worst: + lines.append( + f" {row.decision_id} pred={row.predicted:.3f} selected={row.selected} " + f"observed={row.observed} nll={row.nll:.3f}" + ) + return "\n".join(lines) + + +def dream(log: DecisionLog) -> DreamReport: + report = DreamReport() + for row in log.read(): + probs = row.get("predicted_probs") or {} + observed = row.get("observed_outcome") + selected = row.get("selected") + if observed is None or selected is None or observed not in probs: + report.skipped += 1 + continue + predicted = float(probs[observed]) + if predicted <= 0: + predicted = 1e-9 + nll = -math.log(predicted) + weight = HUMAN_WEIGHT if row.get("label_source") == "human" else SELF_WEIGHT + if row.get("label_source") == "human": + report.human_overrides += 1 + report.rows.append( + RowCost( + decision_id=row["id"], + predicted=predicted, + observed=observed, + selected=selected, + nll=nll, + weight=weight, + correct=(selected == observed), + ) + ) + return report \ No newline at end of file diff --git a/semif_agent/engine.py b/semif_agent/engine.py new file mode 100644 index 0000000..00433d6 --- /dev/null +++ b/semif_agent/engine.py @@ -0,0 +1,93 @@ +"""Real SemIf decision engine (no mocking). + +Wraps the SemIf package (`semif_phase1`). The import and model load happen +lazily so the rest of the agent is pure stdlib and testable without SemIf +installed. On hardware that is neither CUDA nor Apple, use the shipped +llama.cpp backend: a local GGUF scored on CPU (or Vulkan if your llama.cpp +build enables it). This engine is only usable on a machine with SemIf and the +pinned GGUF available; elsewhere calls raise EngineUnavailable. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + +from .decisions import DecisionRequest, DecisionResult + + +class EngineUnavailable(RuntimeError): + """SemIf is not installed or the configured model is missing.""" + + +@dataclass +class EngineConfig: + backend: str = "llamacpp" + source: str = "Qwen/Qwen3.5-4B" + revision: str = "" + gguf: str = "" + context_tokens: int = 4096 + threads: int | None = None + + +class SemIfEngine: + """One pinned SemIf model, loaded once and used for every decision.""" + + def __init__(self, config: EngineConfig): + self.config = config + self._model = None + self._tokenizer = None + self._metadata = None + + def _ensure_loaded(self) -> tuple: + if self._model is not None: + return self._model, self._tokenizer, self._metadata + if self.config.backend != "llamacpp": + raise EngineUnavailable( + f"Unsupported backend {self.config.backend!r}; use 'llamacpp'." + ) + if not self.config.gguf or not Path(self.config.gguf).is_file(): + raise EngineUnavailable( + f"GGUF not found: {self.config.gguf!r}. Set engine.gguf in config.json." + ) + try: + from semif_phase1 import llamacpp_backend as backend + except ImportError as exc: + raise EngineUnavailable( + "SemIf is not installed here. Install it on the target box with " + "`pip install -e '.[test,llamacpp]'`." + ) from exc + try: + model, tokenizer, metadata = backend.load_model( + self.config.source, + self.config.revision, + self.config.gguf, + threads=self.config.threads, + context_tokens=self.config.context_tokens, + ) + except Exception as exc: + raise EngineUnavailable(f"Failed to load the SemIf model: {exc}") from exc + self._model, self._tokenizer, self._metadata = model, tokenizer, metadata + return model, tokenizer, metadata + + @property + def loaded(self) -> bool: + return self._model is not None + + def call(self, request: DecisionRequest) -> DecisionResult: + model, tokenizer, metadata = self._ensure_loaded() + from semif_phase1 import llamacpp_backend as backend + + row = request.to_semif_row() + result = backend.score(model, tokenizer, row, metadata) + return DecisionResult( + request=request, + option_ids=list(result["option_ids"]), + probabilities=list(result["probabilities"]), + extra={ + "prompt_sha256": result.get("prompt_sha256"), + "input_tokens": result.get("input_tokens"), + "forward_seconds": result.get("forward_seconds"), + "total_seconds": result.get("total_seconds"), + }, + ) diff --git a/semif_agent/llm.py b/semif_agent/llm.py new file mode 100644 index 0000000..da16849 --- /dev/null +++ b/semif_agent/llm.py @@ -0,0 +1,86 @@ +"""Self-assessment LLM client. + +Talks to a local OpenAI-compatible server (e.g. ollama, llama.cpp server) for +the observe -> assess phase of the skill loop. Real, not mocked; the endpoint +must be reachable. Uses only the stdlib HTTP client so the core stays +dependency-free. +""" + +from __future__ import annotations + +import json +import urllib.error +import urllib.request +from dataclasses import dataclass, field +from typing import Any + + +@dataclass +class Assessment: + success: bool + summary: str + updated_request: str | None = None + extra: dict[str, Any] = field(default_factory=dict) + + +class LLMClient: + def __init__(self, base_url: str, model: str, timeout: float = 120.0): + self.base_url = base_url.rstrip("/") + self.model = model + self.timeout = timeout + + def _chat(self, messages: list[dict], temperature: float = 0.0) -> str: + url = f"{self.base_url}/chat/completions" + body = json.dumps( + {"model": self.model, "messages": messages, "temperature": temperature} + ).encode("utf-8") + request = urllib.request.Request( + url, data=body, headers={"Content-Type": "application/json"} + ) + try: + with urllib.request.urlopen(request, timeout=self.timeout) as response: + payload = json.loads(response.read().decode("utf-8")) + except urllib.error.URLError as exc: + raise RuntimeError( + f"LLM endpoint unreachable at {url}: {exc}. Is your local server running?" + ) from exc + return payload["choices"][0]["message"]["content"] + + def assess(self, skill: str, request_text: str, action_log: str) -> Assessment: + system = ( + "You are the self-assessment step of an agent skill run. Decide whether " + "the skill achieved its goal. Reply with JSON only: " + '{"success": true|false, "summary": "", "updated_request": ' + '""}. success is true only if the goal was met.' + ) + user = ( + f"Skill: {skill}\n" + f"Goal request: {request_text}\n" + f"What was done:\n{action_log}\n" + ) + try: + raw = self._chat( + [ + {"role": "system", "content": system}, + {"role": "user", "content": user}, + ] + ) + parsed = self._parse_json(raw) + except Exception as exc: + return Assessment(success=False, summary=f"assessment failed: {exc}") + success = bool(parsed.get("success")) + summary = str(parsed.get("summary", "")) + updated = parsed.get("updated_request") + return Assessment( + success=success, + summary=summary, + updated_request=None if updated is None else str(updated), + ) + + @staticmethod + def _parse_json(raw: str) -> dict: + text = raw.strip() + start, end = text.find("{"), text.rfind("}") + if start != -1 and end != -1: + text = text[start : end + 1] + return json.loads(text) diff --git a/semif_agent/log.py b/semif_agent/log.py new file mode 100644 index 0000000..2072bac --- /dev/null +++ b/semif_agent/log.py @@ -0,0 +1,78 @@ +"""Decision logging: every SemIf call is written as a labeled training row. + +Rows match the SemIf `decisions.jsonl` shape plus the prediction-observation +cost fields. `observed_outcome` is the label for the cost function; by default +it is the option that was actually selected (self-consistent), and a human +override can relabel a row to the correct outcome with a higher weight. +""" + +from __future__ import annotations + +import json +import time +from pathlib import Path + +from .decisions import DecisionRequest, DecisionResult + + +class DecisionLog: + def __init__(self, path: str = "data/decisions.jsonl"): + self.path = Path(path) + + def append( + self, + request: DecisionRequest, + result: DecisionResult, + label: str | None = None, + extra: dict | None = None, + ) -> None: + """Append one decision row. `label` overrides observed_outcome.""" + observed = label if label is not None else result.selected + source = "human" if label is not None else "self" + row = { + "id": request.id, + "ts": time.time(), + "state": request.state, + "question": request.question, + "options": [{"id": o.id, "description": o.description} for o in request.options], + "predicted_probs": result.probs, + "selected": result.selected, + "observed_outcome": observed, + "label_source": source, + } + if extra: + row["extra"] = extra + self.path.parent.mkdir(parents=True, exist_ok=True) + with self.path.open("a") as handle: + handle.write(json.dumps(row) + "\n") + + def relabel(self, decision_id: str, observed: str) -> bool: + """Human override: set a corrected observed outcome for one row.""" + rows = self.read() + found = False + for row in rows: + if row["id"] == decision_id: + row["observed_outcome"] = observed + row["label_source"] = "human" + found = True + if not found: + return False + self._write(rows) + return True + + def read(self) -> list[dict]: + if not self.path.is_file(): + return [] + rows = [] + with self.path.open("r") as handle: + for line in handle: + line = line.strip() + if line: + rows.append(json.loads(line)) + return rows + + def _write(self, rows: list[dict]) -> None: + self.path.parent.mkdir(parents=True, exist_ok=True) + with self.path.open("w") as handle: + for row in rows: + handle.write(json.dumps(row) + "\n") diff --git a/semif_agent/queue.py b/semif_agent/queue.py new file mode 100644 index 0000000..d78e00c --- /dev/null +++ b/semif_agent/queue.py @@ -0,0 +1,83 @@ +"""Urgency priority queue. + +Items are sorted by descending urgency weight, then FIFO arrival, then most +recent first. Queued items age upward so they cannot starve, and the queue has +a maximum depth. +""" + +from __future__ import annotations + +import heapq +import itertools +import time +from dataclasses import dataclass +from typing import Iterator + +from .decisions import Request + +MIN_RECENCY = 0.0 + + +@dataclass +class QueueItem: + weight: float + seq: int + recency: float + request: Request + + def key(self) -> tuple: + return (-self.weight, self.seq) + + +class UrgencyQueue: + """A max-by-urgency priority queue with FIFO/recency tie-breaking.""" + + def __init__(self, max_size: int = 100, age_rate: float = 0.0): + self.max_size = max_size + self.age_rate = age_rate + self._entries: list[tuple] = [] + self._counter = itertools.count() + + def push(self, request: Request, weight: float, recency: float | None = None) -> bool: + """Insert an item. Returns False if the queue is full.""" + if len(self._entries) >= self.max_size: + return False + item = QueueItem( + weight=weight, + seq=next(self._counter), + recency=recency if recency is not None else time.time(), + request=request, + ) + entry = (*item.key(), next(self._counter), item) + heapq.heappush(self._entries, entry) + return True + + def pop(self) -> Request | None: + if not self._entries: + return None + return heapq.heappop(self._entries)[-1].request + + def peek(self) -> Request | None: + if not self._entries: + return None + return self._entries[0][-1].request + + def age(self, dt: float = 1.0) -> None: + """Pull queued priorities toward critical so old items catch up.""" + if self.age_rate <= 0: + return + rebuilt = [] + for entry in self._entries: + item = entry[-1] + item.weight += self.age_rate * dt * (1.0 - item.weight) + rebuilt.append((*item.key(), next(self._counter), item)) + self._entries = rebuilt + heapq.heapify(self._entries) + + def __len__(self) -> int: + return len(self._entries) + + def items(self) -> Iterator[tuple[float, Request]]: + for entry in sorted(self._entries, key=lambda e: e[:-1]): + item = entry[-1] + yield item.weight, item.request \ No newline at end of file diff --git a/semif_agent/scheduler.py b/semif_agent/scheduler.py new file mode 100644 index 0000000..fd87721 --- /dev/null +++ b/semif_agent/scheduler.py @@ -0,0 +1,212 @@ +"""The scheduler: gate, choice, score, queue, dispatch. + +Every SemIf decision (gate, choice, score, navigation, prediction) is logged. +A high-priority input can preempt the current process, which requeues with its +state preserved; a deferred input is scored and queued by urgency. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +from .decisions import DecisionRequest, Option, Request +from .engine import SemIfEngine +from .llm import LLMClient +from .log import DecisionLog +from .queue import UrgencyQueue +from .skill import SkillRunner +from .skills import ( + ActionContext, + CreateSkill, + Skill, + build_skills, + build_tree, + compose_state, + navigate, +) + +GATE_YES = "yes" +CHOICE_INTERRUPT = "interrupt" +URGENCY_OPTIONS = [ + ("critical", "Immediate danger or critical failure."), + ("high", "Important but not dangerous."), + ("medium", "Should be handled reasonably soon."), + ("low", "Can wait."), +] +URGENCY_WEIGHTS = {"critical": 1.0, "high": 0.75, "medium": 0.5, "low": 0.25} + + +@dataclass +class Process: + request: Request + skill: str + weight: float + + +@dataclass +class DispatchResult: + kind: str # ran | create_skill | error + summary: str + skill: str | None = None + decisions_logged: int = 0 + + +class Scheduler: + def __init__( + self, + engine: SemIfEngine, + llm: LLMClient, + log: DecisionLog, + config: dict, + tau: float = 0.6, + max_reentries: int = 3, + ): + self.engine = engine + self.llm = llm + self.log = log + self.config = config + self.tau = tau + self.max_reentries = max_reentries + self.queue = UrgencyQueue( + max_size=int(config.get("queue", {}).get("max_size", 100)), + age_rate=float(config.get("queue", {}).get("age_rate", 0.0)), + ) + self.skills = build_skills(config) + self.tree = build_tree(self.skills) + self.ctx = ActionContext(engine=self.engine, config=config) + self.runner = SkillRunner(self.ctx, self.llm, self.log) + self.current: Process | None = None + + # ---- decision templates (all real SemIf, all logged) ---- + + def _contains_request(self, request: Request) -> bool: + decision = DecisionRequest( + state=compose_state(request), + question="Does this input contain an actionable request?", + options=[Option(GATE_YES, "Yes, it is actionable."), Option("no", "No, it is not.")], + ) + result = self.engine.call(decision) + self.log.append(decision, result, extra={"phase": "gate"}) + return result.prob(GATE_YES) >= self.tau + + def _choice(self, request: Request, current: Process) -> bool: + decision = DecisionRequest( + state=compose_state(request, current=current.skill), + question="Should this be allowed to interrupt the current process?", + options=[ + Option(CHOICE_INTERRUPT, "Yes, interrupt the current process."), + Option("defer", "No, wait until the current process finishes."), + ], + ) + result = self.engine.call(decision) + self.log.append(decision, result, extra={"phase": "choice", "current": current.skill}) + return result.prob(CHOICE_INTERRUPT) >= self.tau + + def _score(self, request: Request, current: str | None = None) -> tuple[float, str]: + decision = DecisionRequest( + state=compose_state(request, current=current), + question="How urgent is this request?", + options=[Option(option_id, description) for option_id, description in URGENCY_OPTIONS], + ) + result = self.engine.call(decision) + self.log.append(decision, result, extra={"phase": "score"}) + label = result.selected + return URGENCY_WEIGHTS[label], label + + # ---- intake ---- + + def submit(self, text: str, source: str = "typed") -> tuple[str, str]: + """Feed one input. Returns (status, detail).""" + from .engine import EngineUnavailable + + try: + return self._submit(text, source) + except EngineUnavailable as exc: + return "error", f"decision engine unavailable: {exc}" + + def _submit(self, text: str, source: str = "typed") -> tuple[str, str]: + request = Request(text, source=source) + if not self._contains_request(request): + return "dropped", "no actionable request" + + if self.current is None: + weight, label = self._score(request) + self.current = Process(request=request, skill="(scheduling)", weight=weight) + outcome = self._dispatch(request) + self.current = None + return "running", f"[{label}] {outcome.summary}" + + interrupt = self._choice(request, self.current) + if interrupt: + previous = self.current + previous.request.resume["from_skill"] = previous.skill + self.queue.push(previous.request, previous.weight) + self.current = Process(request=request, skill="(scheduling)", weight=1.0) + outcome = self._dispatch(request) + self.current = None + return "preempted", f"interrupted {previous.skill}; {outcome.summary}" + + weight, label = self._score(request, current=self.current.skill) + ok = self.queue.push(request, weight) + if not ok: + return "rejected", "queue is full" + return "queued", f"urgency {label} (weight {weight:.2f})" + + def busy(self, text: str, skill: str = "(driving)") -> None: + """Set a fake in-progress process so the choice/score path is exercised.""" + self.current = Process(request=Request(text, source="busy"), skill=skill, weight=1.0) + + def idle(self) -> None: + self.current = None + + def run_queue(self) -> list[tuple[str, str]]: + """Process the queue while idle. Returns the outcomes.""" + from .engine import EngineUnavailable + + results = [] + while self.current is None and len(self.queue) > 0: + request = self.queue.pop() + self.current = Process(request=request, skill="(scheduling)", weight=0.0) + try: + outcome = self._dispatch(request) + except EngineUnavailable as exc: + outcome = DispatchResult(kind="error", summary=f"engine unavailable: {exc}") + self.current = None + results.append(("ran", f"[{request.id}] {outcome.summary}")) + return results + + # ---- dispatch ---- + + def _dispatch(self, request: Request) -> DispatchResult: + navigation = navigate(self.engine, request, self.tree) + if isinstance(navigation, CreateSkill): + return DispatchResult( + kind="create_skill", + summary="skill authoring via opencode is deferred to v2; request logged.", + ) + outcome = self.runner.run(navigation, request) + if outcome.error: + return DispatchResult(kind="error", summary=f"skill error: {outcome.error}") + if outcome.updated_request and request.reentries < self.max_reentries: + self.queue.push(_requeue(request, outcome.updated_request), 0.5) + return DispatchResult( + kind="ran", + summary=f"{navigation.name}: {'ok' if outcome.success else 'failed'} — {outcome.summary}", + skill=navigation.name, + decisions_logged=outcome.decisions_logged, + ) + + def status(self) -> str: + lines = [] + current = f"{self.current.skill} ({self.current.request.id})" if self.current else "idle" + lines.append(f"current: {current}") + lines.append(f"queue: {len(self.queue)} pending") + for weight, request in self.queue.items(): + lines.append(f" {request.id} w={weight:.2f} {request.text[:60]}") + return "\n".join(lines) + + +def _requeue(request: Request, updated_text: str) -> Request: + updated = Request(updated_text, source="requeue") + updated.reentries = request.reentries + 1 + return updated \ No newline at end of file diff --git a/semif_agent/skill.py b/semif_agent/skill.py new file mode 100644 index 0000000..ed9759c --- /dev/null +++ b/semif_agent/skill.py @@ -0,0 +1,65 @@ +"""The skill execution loop: observe -> predict -> act -> observe -> assess. + +Every SemIf decision made during a run is logged as a training row; the +assessment outcome is kept on the row so the dream pass can weigh failed runs. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +from .decisions import Request +from .llm import Assessment, LLMClient +from .log import DecisionLog +from .skills import ActionContext, Prediction, Skill + + +@dataclass +class RunResult: + skill: str + success: bool + summary: str + action_log: str + new_state: str + updated_request: str | None = None + decisions_logged: int = 0 + error: str | None = None + + +class SkillRunner: + def __init__(self, ctx: ActionContext, llm: LLMClient, log: DecisionLog): + self.ctx = ctx + self.llm = llm + self.log = log + + def run(self, skill: Skill, request: Request) -> RunResult: + baseline = request.text + try: + prediction = skill.predict(self.ctx, request) if skill.predict else Prediction(text="") + action = skill.act(self.ctx, request, prediction) + observed = action.new_state + assessment: Assessment = self.llm.assess(skill.name, baseline, action.action_log) + except Exception as exc: + return RunResult( + skill=skill.name, + success=False, + summary="", + action_log="", + new_state=baseline, + error=str(exc), + ) + + run_ok = assessment.success + decisions = getattr(prediction, "decisions", []) + for decision, result in decisions: + self.log.append(decision, result, extra={"skill": skill.name, "run_ok": run_ok}) + + return RunResult( + skill=skill.name, + success=assessment.success, + summary=assessment.summary, + action_log=action.action_log, + new_state=observed, + updated_request=assessment.updated_request, + decisions_logged=len(decisions), + ) \ No newline at end of file diff --git a/semif_agent/skills.py b/semif_agent/skills.py new file mode 100644 index 0000000..3b19d2e --- /dev/null +++ b/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) diff --git a/tests/integration/test_pipeline.py b/tests/integration/test_pipeline.py new file mode 100644 index 0000000..7263690 --- /dev/null +++ b/tests/integration/test_pipeline.py @@ -0,0 +1,79 @@ +"""End-to-end pipeline test. Run ONLY on the box with real SemIf + real LLM. + + python -m pytest tests/integration -q + +The decision engine is real (llamacpp GGUF) and self-assessment is a real +local LLM endpoint. If either is unavailable this fails loudly — no mocking. +""" + +import json +from pathlib import Path + +import pytest + +from semif_agent.cli import build_scheduler, load_config +from semif_agent.dream import dream +from semif_agent.engine import EngineUnavailable + + +def require_real(config: dict): + from semif_agent.engine import SemIfEngine, EngineConfig + + engine = SemIfEngine( + EngineConfig( + backend=config.get("engine", {}).get("backend", "llamacpp"), + source=config.get("engine", {}).get("source", ""), + revision=config.get("engine", {}).get("revision", ""), + gguf=config.get("engine", {}).get("gguf", ""), + context_tokens=int(config.get("engine", {}).get("context_tokens", 4096)), + threads=config.get("engine", {}).get("threads"), + ) + ) + try: + engine._ensure_loaded() + except EngineUnavailable as exc: + pytest.fail(f"real engine unavailable: {exc}") + + +def test_pipeline_end_to_end(tmp_path): + config = load_config() + require_real(config) + config["log"] = str(tmp_path / "decisions.jsonl") + scheduler, config = build_scheduler(config) + + inputs = [ + "send my girlfriend an email that says I'm going to be late to the party", + "tell me if my package was delivered", + ] + for text in inputs: + status, detail = scheduler.submit(text) + print(f"[{status}] {detail}") + assert status in ("running", "preempted", "queued", "rejected") + + scheduler.run_queue() + + rows = scheduler.log.read() + assert len(rows) > 0, "expected SemIf decisions to be logged" + + report = dream(scheduler.log) + assert report.cross_entropy is not None + print(report.render()) + + skills = scheduler.status() + assert "queue:" in skills + + contacts_path = Path(config.get("skills", {}).get("contacts", "data/contacts.json")) + assert contacts_path.is_file() + + +def test_busy_choice_path(tmp_path): + config = load_config() + require_real(config) + config["log"] = str(tmp_path / "decisions.jsonl") + scheduler, config = build_scheduler(config) + + scheduler.busy("driving on the freeway", skill="driving") + status, detail = scheduler.submit("send my girlfriend an email that says I'm going to be late") + print(f"[{status}] {detail}") + assert status in ("preempted", "queued", "dropped") + scheduler.idle() \ No newline at end of file diff --git a/tests/test_contract.py b/tests/test_contract.py new file mode 100644 index 0000000..8113984 --- /dev/null +++ b/tests/test_contract.py @@ -0,0 +1,42 @@ +from semif_agent.decisions import DecisionRequest, DecisionResult, Option, Request + + +def test_to_semif_row_shape(): + request = DecisionRequest( + state="some state", + question="some question?", + options=[Option("a", "A."), Option("b", "B.")], + id="abc", + ) + row = request.to_semif_row() + assert row["id"] == "abc" + assert row["state"] == "some state" + assert row["question"] == "some question?" + assert row["options"] == [ + {"id": "a", "description": "A."}, + {"id": "b", "description": "B."}, + ] + + +def test_result_selected_and_probs(): + request = DecisionRequest( + state="s", + question="q", + options=[Option("a", "A."), Option("b", "B.")], + ) + result = DecisionResult( + request=request, option_ids=["a", "b"], probabilities=[0.3, 0.7] + ) + assert result.selected == "b" + assert result.probs == {"a": 0.3, "b": 0.7} + assert result.prob("a") == 0.3 + + +def test_request_requeue_preserves_state(): + original = Request(text="t", priority=0.7) + original.resume["from_skill"] = "email.compose" + updated = original.copy_for_requeue() + assert updated.id == original.id + assert updated.priority == original.priority + assert updated.resume["from_skill"] == "email.compose" + assert updated.reentries == original.reentries + 1 \ No newline at end of file diff --git a/tests/test_dream.py b/tests/test_dream.py new file mode 100644 index 0000000..9eaec81 --- /dev/null +++ b/tests/test_dream.py @@ -0,0 +1,89 @@ +import math + +import pytest + +from semif_agent.decisions import DecisionRequest, DecisionResult, Option +from semif_agent.dream import HUMAN_WEIGHT, SELF_WEIGHT, dream +from semif_agent.log import DecisionLog + + +def make_result(probs): + request = DecisionRequest( + state="state", + question="question", + options=[Option(key, key) for key in probs], + ) + result = DecisionResult( + request=request, + option_ids=list(probs), + probabilities=list(probs.values()), + ) + return request, result + + +def test_cross_entropy_matches_hand_calculation(tmp_path): + log = DecisionLog(str(tmp_path / "decisions.jsonl")) + probs = {"yes": 0.8, "no": 0.2} + request, result = make_result(probs) + log.append(request, result, label="no") + report = dream(log) + assert report.cross_entropy == pytest.approx(-math.log(0.2)) + assert report.rows[0].nll == pytest.approx(-math.log(0.2)) + assert report.rows[0].weight == HUMAN_WEIGHT + + +def test_default_label_is_selected(tmp_path): + log = DecisionLog(str(tmp_path / "decisions.jsonl")) + request, result = make_result({"yes": 0.9, "no": 0.1}) + log.append(request, result) + report = dream(log) + row = report.rows[0] + assert row.observed == "yes" + assert row.correct is True + assert row.weight == SELF_WEIGHT + + +def test_accuracy_and_ece(tmp_path): + log = DecisionLog(str(tmp_path / "decisions.jsonl")) + for probs, label in [ + ({"yes": 0.9, "no": 0.1}, "yes"), + ({"yes": 0.6, "no": 0.4}, "no"), + ({"yes": 0.9, "no": 0.1}, "yes"), + ]: + request, result = make_result(probs) + log.append(request, result, label=label) + report = dream(log) + assert report.accuracy == pytest.approx(2 / 3) + assert report.ece is not None and 0.0 <= report.ece <= 1.0 + + +def test_relabel_human_override(tmp_path): + log = DecisionLog(str(tmp_path / "decisions.jsonl")) + request, result = make_result({"yes": 0.9, "no": 0.1}) + log.append(request, result) + assert log.relabel(request.id, "no") is True + rows = log.read() + assert rows[0]["observed_outcome"] == "no" + assert rows[0]["label_source"] == "human" + assert log.relabel("missing", "yes") is False + + +def test_skips_unlabeled(tmp_path): + log = DecisionLog(str(tmp_path / "decisions.jsonl")) + log.path.parent.mkdir(parents=True, exist_ok=True) + log.path.write_text('{"id": "x", "predicted_probs": {"a": 1.0}}\n') + report = dream(log) + assert report.skipped == 1 + assert report.rows == [] + assert report.cross_entropy is None + + +def test_clamped_probability_never_zero(tmp_path): + log = DecisionLog(str(tmp_path / "decisions.jsonl")) + log.path.parent.mkdir(parents=True, exist_ok=True) + log.path.write_text( + '{"id": "x", "predicted_probs": {"a": 0.0, "b": 1.0}, ' + '"selected": "b", "observed_outcome": "a", "label_source": "human"}\n' + ) + report = dream(log) + assert report.rows[0].nll == pytest.approx(-math.log(1e-9)) \ No newline at end of file diff --git a/tests/test_queue.py b/tests/test_queue.py new file mode 100644 index 0000000..149024f --- /dev/null +++ b/tests/test_queue.py @@ -0,0 +1,74 @@ +import pytest + +from semif_agent.queue import UrgencyQueue +from semif_agent.decisions import Request + + +def make(text, weight): + return Request(text=text), weight + + +def test_order_by_weight_desc(): + q = UrgencyQueue() + req_a, _ = make("low", 0.25) + req_c, _ = make("critical", 1.0) + req_h, _ = make("high", 0.75) + q.push(req_a, 0.25) + q.push(req_c, 1.0) + q.push(req_h, 0.75) + assert q.pop().text == "critical" + assert q.pop().text == "high" + assert q.pop().text == "low" + assert q.pop() is None + + +def test_fifo_tiebreak(): + q = UrgencyQueue() + req_a, _ = make("first", 0.5) + req_b, _ = make("second", 0.5) + q.push(req_a, 0.5) + q.push(req_b, 0.5) + assert q.pop().text == "first" + assert q.pop().text == "second" + + +def test_fifo_beats_recency(): + q = UrgencyQueue() + old, _ = make("older", 0.5) + new, _ = make("newer", 0.5) + q.push(old, 0.5, recency=1.0) + q.push(new, 0.5, recency=2.0) + assert q.pop().text == "older" + + +def test_peek_does_not_remove(): + q = UrgencyQueue() + req, _ = make("peek", 0.9) + q.push(req, 0.9) + assert q.peek().text == "peek" + assert len(q) == 1 + + +def test_bounds_reject(): + q = UrgencyQueue(max_size=2) + assert q.push(make("a", 1.0)[0], 1.0) + assert q.push(make("b", 1.0)[0], 1.0) + assert not q.push(make("c", 1.0)[0], 1.0) + + +def test_age_raises_priority(): + q = UrgencyQueue(age_rate=1.0) + old, _ = make("aging", 0.2) + new, _ = make("busy", 1.0) + q.push(old, 0.2) + q.push(new, 1.0) + q.age(dt=10.0) + assert q.peek().text == "aging" + + +def test_items_sorted(): + q = UrgencyQueue() + q.push(make("b", 0.5)[0], 0.5) + q.push(make("a", 1.0)[0], 1.0) + weights = [weight for weight, _ in q.items()] + assert weights == [1.0, 0.5] \ No newline at end of file