feat: learning loops - price list, lessons from corrections, what it learns, eval candidates (09)

core/src/ooat_core/conversation_endpoints.py

@@ -26,7 +26,7 @@
 SSE_POLL_S, SSE_MAX_S = 1.0, 900.0
 CONVERSATION = r"^cnv_[0-7][0-9A-HJKMNP-TV-Z]{25}$"
 MEMORY = r"^mem_[0-7][0-9A-HJKMNP-TV-Z]{25}$"
-TAP = r"^(yes|differently|new_thing|make_task|todo|(forget|done):mem_[0-7][0-9A-HJKMNP-TV-Z]{25})$"
+TAP = r"^(yes|differently|new_thing|make_task|todo|remember|(forget|done):mem_[0-7][0-9A-HJKMNP-TV-Z]{25})$"
 
 
 class _Body(BaseModel):
@@ -107,8 +107,12 @@ def send(body: MessageIn, request: Request, who: Principal = Depends(scope("subm
     @api.post("/conversations/{conversation}/taps")
     def tap(conversation: str, body: TapIn, request: Request, who: Principal = Depends(scope("answer"))) -> dict:
         server = server_of(request)
-        found, _ = own_conversation(server, who, conversation)
+        found, events = own_conversation(server, who, conversation)
         require_within(who, project_class(server, found.project))
+        if body.tap in ("remember", "todo"):  # the item keeps the class of the decision it comes from
+            last = conversations.last_decision(conversations.events_of(events, conversation))
+            if last is not None:
+                require_within(who, last["body"]["data_class"])
         if body.tap.startswith(("forget:", "done:")):
             mem_id = body.tap.split(":", 1)[1]
             events = server.runtime().ledger.events(types=memory.EVENTS)

core/src/ooat_core/endpoints.py

@@ -20,7 +20,7 @@
 from fastapi.responses import JSONResponse, Response, StreamingResponse
 from pydantic import BaseModel, ConfigDict, Field, StrictInt, StrictStr
 
-from . import connector_admin, hil, stats, views
+from . import connector_admin, hil, learning, stats, views
 from .api import ApiError, Principal, Server, principal, require_within, scope, server_of
 from .gate import settings_from_config
 from .ids import parse_artifact_ref
@@ -372,6 +372,15 @@ def approve_hooks(connector_id: str, body: HooksIn, request: Request,
 
     # Usage -----------------------------------------------------------------------------------------------------
 
+    @api.get("/learning")
+    def learning_view(request: Request, who: Principal = Depends(scope("read"))) -> dict:
+        """What it learns (plan 09 decision 7): counts over the whole ledger, task ids and notes within the cap."""
+        server = server_of(request)
+        tasks = submitted_tasks(server)
+        visible = {t for t, c in task_classes(server, tasks).items() if views.within(c, who.max_data_class)}
+        return learning.summary(server.runtime().ledger.events(), visible=visible.__contains__,
+                                reception_interim=server.config.reception.theta)
+
     @api.get("/stats")
     def usage(request: Request, period: str = "7d", project: str | None = None, group: str = "role",
               who: Principal = Depends(scope("read"))) -> dict:

core/src/ooat_core/gate.py

@@ -15,6 +15,7 @@
 from typing import Literal
 
 from .acceptance import CRITIC_SYSTEM, CRITIC_TIER
+from . import learning
 from .config import Config
 from .connectors import DecisionQuestion, DecisionRequest, ModelRequest
 from .gateway import Gateway, GatewayError
@@ -26,6 +27,7 @@
 from .worker import PREVIEW_CHARS, worker_system
 
 ACTOR = {"kind": "system", "id": "ooat-gate"}
+WORKER_CAPABILITY = "cap.general.complete_task"  # runtime.CAPABILITY; the runtime imports this module
 WORKER_TIER = "workhorse"
 # A task that does not state its risk class is judged by the stricter R1 tolerance, never the laxer R0
 # (ADR 0011: below θ the safer outcome applies).
@@ -258,10 +260,16 @@ def run(self, task: str) -> GateOutcome:
             missing = f"no permitted route for {data_class} data: {error.message}"
             self._decided(task, "T0", rules, [{"topology": "T2", "eliminated_by": rules or ["A10"]}], decided)
             return self._close(task, data_class, budget, "CLOSED_ABSTAINED", "Not run: no permitted route.", missing)
-        candidates = [{"topology": "T2", "model_usd": estimate}]
-        if value is not None and estimate > value:
+        p = learning.p_accept(self._ledger.events(types=list(learning.EVENTS)), WORKER_CAPABILITY,
+                              submitted["body"].get("task_type"), learning.card_prior(WORKER_CAPABILITY))
+        candidate = {"topology": "T2", "model_usd": estimate, "p_accept": round(p, 4)}
+        if value is not None:  # spec §6 with no HIL or silent-error term measured yet (plan 09 decision 4)
+            candidate["eu_usd"] = value * p - estimate
+        candidates = [candidate]
+        if value is not None and candidate["eu_usd"] < 0:
             self._decided(task, "T0", rules, candidates, decided)
-            missing = f"the estimated cost {estimate:.4f} USD exceeds the task value {value:.2f} USD"
+            missing = (f"the estimated cost {estimate:.4f} USD exceeds the task's expected value {value * p:.4f} USD "
+                       f"(value {value:.2f} USD × acceptance {p:.2f})")
             return self._close(task, data_class, budget, "CLOSED_ABSTAINED", "Not run: not worth its cost.", missing,
                                estimate)
 

core/src/ooat_core/learning.py

@@ -0,0 +1,171 @@
+"""What OOAT learns from the ledger (plan 09): the learning price list of spec §11 and the What it learns figures.
+
+Nothing is stored. Like θ (thresholds.py), every figure is recomputed from the events on each read, so the card in
+the repository keeps `"observed": {"n": 0}` and `cost_card.observed` changes only through ooat-core (plan 09
+decision 1).
+"""
+
+from collections.abc import Callable, Iterable
+from dataclasses import asdict, dataclass
+
+from .catalog import load_card
+from .pii import scan
+from .thresholds import (RECEPTION_POINT, quick_cancels, rated_decisions, reception_threshold, reception_verdicts,
+                         threshold)
+
+EVENTS = ("TASK_SUBMITTED", "CONTRACT_ISSUED", "TASK_CLOSED", "TASK_RATED")
+GENERAL = "general"  # the task type of a task submitted without one
+PRIOR_WEIGHT = 5  # the card's prior counts as five observations (spec §11)
+ABSTAINED = "CLOSED_ABSTAINED"
+REPLACED = ("escalated", "retried")  # calls whose decision the next call replaced: the person never saw them
+ATTENTION = 10  # tasks listed as needing a look
+
+
+@dataclass(frozen=True)
+class Observed:
+    capability: str
+    task_type: str
+    n: int  # rated runs counted in p_accept
+    accepted: int
+    prior: float
+    p_accept: float  # the Beta mean
+    runs: int  # closed runs, rated or not
+    p_abstain: float | None
+
+    def as_dict(self) -> dict:
+        return asdict(self) | {"p_silent": None}  # needs DEFECT_FOUND (spec §11), not recorded yet
+
+
+def card_prior(capability: str) -> float:
+    """The card's prior.p_accept; 0.5 when the card states none."""
+    return load_card("capabilities", capability).get("cost_card", {}).get("prior", {}).get("p_accept", 0.5)
+
+
+def _runs(events: Iterable[dict]) -> list[tuple[str, str, str, bool | None]]:
+    """(capability, task type, closing state, accepted or None) of every closed task that ran a contract."""
+    types, capability, closed, rated = {}, {}, {}, {}
+    for event in events:
+        task, body = event.get("task"), event["body"]
+        if event["type"] == "TASK_SUBMITTED":
+            types[task] = body.get("task_type") or GENERAL
+        elif event["type"] == "CONTRACT_ISSUED":
+            capability.setdefault(task, body["contract"]["capability"])
+        elif event["type"] == "TASK_CLOSED":
+            closed[task] = body["state"]
+        elif event["type"] == "TASK_RATED":
+            rated[task] = body["accepted"]
+    return [(capability[t], types.get(t, GENERAL), state, rated.get(t)) for t, state in closed.items()
+            if t in capability]
+
+
+def _counted(state: str, accepted: bool | None) -> bool | None:
+    """A success, a failure, or None when the run does not count (plan 09 decision 2): unrated, or an abstention the
+    operator accepted as justified."""
+    if accepted is None or (state == ABSTAINED and accepted):
+        return None
+    return accepted
+
+
+def _beta_mean(successes: int, n: int, prior: float) -> float:
+    return (prior * PRIOR_WEIGHT + successes) / (PRIOR_WEIGHT + n)
+
+
+def p_accept(events: Iterable[dict], capability: str, task_type: str | None, prior: float) -> float:
+    """The Beta mean of acceptance for one capability and task type."""
+    wanted = task_type or GENERAL
+    counted = [_counted(state, accepted) for cap, kind, state, accepted in _runs(events)
+               if (cap, kind) == (capability, wanted)]
+    counted = [c for c in counted if c is not None]
+    return _beta_mean(sum(counted), len(counted), prior)
+
+
+def observed(events: Iterable[dict], priors: dict[str, float]) -> list[Observed]:
+    """One row per capability and task type that has closed runs, sorted by both."""
+    groups: dict[tuple[str, str], list[tuple[str, bool | None]]] = {}
+    for cap, kind, state, accepted in _runs(events):
+        groups.setdefault((cap, kind), []).append((state, accepted))
+    rows = []
+    for (cap, kind), runs in sorted(groups.items()):
+        counted = [c for c in (_counted(s, a) for s, a in runs) if c is not None]
+        prior = priors.get(cap, 0.5)
+        mean = _beta_mean(sum(counted), len(counted), prior)
+        rows.append(Observed(cap, kind, len(counted), sum(counted), prior, mean, len(runs),
+                             sum(s == ABSTAINED for s, _ in runs) / len(runs)))
+    return rows
+
+
+
+# What it learns (plan 09 decision 7) ------------------------------------------------------------------------------
+
+def _thresholds(events: list[dict], reception_interim: float) -> list[dict]:
+    rows: dict[tuple[str, str, str], list[bool]] = {}
+    for rated in [*rated_decisions(events), *reception_verdicts(events)]:
+        rows.setdefault((rated.point, rated.engine, rated.model), []).append(rated.correct)
+    found = []
+    for (point, engine, model), correct in sorted(rows.items()):
+        theta = (reception_threshold(events, engine, model, reception_interim) if point == RECEPTION_POINT
+                 else threshold(events, point, engine, model))
+        found.append({"point": point, "engine": engine, "model": model, "rated": len(correct),
+                      "confirmed": sum(correct), "theta": theta})
+    return found
+
+
+def _reception(events: list[dict]) -> tuple[list[dict], dict]:
+    rows: dict[str, dict] = {}
+    last: dict[str, dict] = {}  # conversation -> the row of its newest decision the person saw
+    cancelled = quick_cancels(events)
+    offered = 0
+    taps = set()  # the remember taps; one counts as kept when its item was stored
+    for event in events:
+        body = event["body"]
+        if event["type"] == "RECEPTION_DECIDED" and body["decision"] not in REPLACED:
+            row = rows.setdefault(body["decision"], {"decision": body["decision"], "count": 0, "differently": 0,
+                                                     "cancelled": 0})
+            row["count"] += 1
+            row["cancelled"] += event["id"] in cancelled
+            last[body["conversation"]] = row
+            offered += "lesson" in body
+        elif event["type"] == "CONVERSATION_MESSAGE" and body.get("tap") in ("differently", "remember"):
+            if body["tap"] == "differently" and body["conversation"] in last:
+                last.pop(body["conversation"])["differently"] += 1
+            if body["tap"] == "remember":
+                taps.add(event["id"])
+    kept = sum(1 for e in events if e["type"] == "MEMORY_NOTED" and e["body"].get("source") in taps)
+    return (sorted(rows.values(), key=lambda r: (-r["count"], r["decision"])),
+            {"lessons_offered": offered, "lessons_kept": kept})
+
+
+def _memory(events: list[dict]) -> dict:
+    """Notes the reception stored on its own, and how many of them the operator forgot later."""
+    noted = {e["body"]["memory"] for e in events if e["type"] == "MEMORY_NOTED" and e["actor"]["kind"] == "system"}
+    forgotten = {e["body"]["memory"] for e in events if e["type"] == "MEMORY_FORGOTTEN"} & noted
+    return {"noted": len(noted), "forgotten": len(forgotten)}
+
+
+def _plain(note: str | None) -> str | None:
+    """A rating note is the operator's free text with no class of its own: shown only when it holds no personal
+    data the scan recognises."""
+    return note if note is None or not scan(note) else None
+
+
+def _attention(events: list[dict], visible: Callable[[str], bool]) -> list[dict]:
+    """The latest rejected tasks within the reader's cap, newest first; a justified abstention needs no look."""
+    closed = {e["task"]: e["body"]["state"] for e in events if e["type"] == "TASK_CLOSED"}
+    found = []
+    for event in events:
+        if event["type"] != "TASK_RATED" or event["body"]["accepted"] or not visible(event["task"]):
+            continue
+        state = closed.get(event["task"], "")
+        found.append({"task": event["task"], "state": state,
+                      "why": "lazy_abstention" if state == ABSTAINED else "rejected",
+                      "note": _plain(event["body"].get("note"))})
+    return found[-ATTENTION:][::-1]
+
+
+def summary(events: list[dict], *, visible: Callable[[str], bool], reception_interim: float) -> dict:
+    """The What it learns figures. Counts cover every task; task ids and rating notes only `visible` ones."""
+    caps = {e["body"]["contract"]["capability"] for e in events if e["type"] == "CONTRACT_ISSUED"}
+    reception, lessons = _reception(events)
+    return {"price_list": [r.as_dict() for r in observed(events, {c: card_prior(c) for c in caps})],
+            "thresholds": _thresholds(events, reception_interim), "reception": reception,
+            "memory": _memory(events) | lessons, "attention": _attention(events, visible)}

core/src/ooat_core/reception/__init__.py

@@ -173,11 +173,24 @@ def _apply_tap(self, found, own, message, tap, operator, channel) -> dict:
                 raise ValueError("there is nothing to keep")
             text = last["body"].get("interpretation") or " ".join(
                 conversations.message_text(e) or "" for e in own if e["id"] in last["body"]["messages"])
+            decided["data_class"] = higher_class(decided["data_class"], last["body"]["data_class"])
             item = memory.note(self.ledger, self.store, scope=found.project, kind="todo", text=text,
-                               data_class=message["body"]["data_class"], source=message["id"],
+                               data_class=decided["data_class"], source=message["id"],
                                actor={"kind": "hil", "id": operator}, conversation=found.id, channel=channel)
             decided |= {"decision": "answer", "memory": [item.id], "buttons": [f"done:{item.id}", f"forget:{item.id}"],
                         "reply": reply(f"🧠 Otevřený bod: {item.text}", f"🧠 Open point: {item.text}")}
+        elif tap == "remember":
+            lesson = (last or {}).get("body", {}).get("lesson")
+            if lesson is None or self.store is None:
+                raise ValueError("there is nothing to remember")
+            scope = memory.OPERATOR_SCOPE if lesson.get("scope") == "operator" and lesson["kind"] == "preference" \
+                else found.project
+            decided["data_class"] = higher_class(decided["data_class"], last["body"]["data_class"])
+            item = memory.note(self.ledger, self.store, scope=scope, kind=lesson["kind"], text=lesson["text"],
+                               data_class=decided["data_class"], source=message["id"],
+                               actor={"kind": "hil", "id": operator}, conversation=found.id, channel=channel)
+            decided |= {"decision": "answer", "memory": [item.id], "buttons": [f"forget:{item.id}"],
+                        "reply": reply(f"🧠 Zapamatováno: {item.text}", f"🧠 Remembered: {item.text}")}
         elif tap.startswith(("forget:", "done:")):
             kind, mem_id = tap.split(":", 1)
             self._own_item(found.project, mem_id)
@@ -323,15 +336,16 @@ def decide(self, conversation: str) -> dict | None:
                 projects=sorted(self.projects), budget_tokens=self.settings.context_tokens)
             base = {"conversation": conversation, "project": found.project, "messages": [m["id"] for m in waiting],
                     "context": context.ids, "data_class": data_class}
-            called = self._ask(base, context.text, said, data_class)
+            marked = self._messages(waiting, marked=True)
+            called = self._ask(base, context.text, marked, data_class)
             if called is None:
                 return self._fallback(found, waiting, base, said)
             decision = called.data
             theta = self.theta(called.engine, called.model)
             if decision["decision"] in ("task", "continue_task") and decision["confidence"] < theta:
                 self._record(base | {"decision": "escalated", "options": [], "buttons": []},
                              called, theta)
-                stronger = self._ask(base, context.text, said, data_class, tier=model.ESCALATION_TIER)
+                stronger = self._ask(base, context.text, marked, data_class, tier=model.ESCALATION_TIER)
                 if stronger is not None:
                     called, decision = stronger, stronger.data
                     theta = self.theta(called.engine, called.model)
@@ -371,10 +385,13 @@ def _record(self, body: dict, called, theta, cost: dict | None = None) -> dict:
         return self.ledger.append(new_event("RECEPTION_DECIDED", task=None, actor=ACTOR, body=body,
                                             cost=cost if called is None else called.cost))
 
-    def _messages(self, waiting: list[dict]) -> str:
+    def _messages(self, waiting: list[dict], marked: bool = False) -> str:
+        """The waiting messages with their attachments' previews; `marked` names a Differently correction as one,
+        for the model (plan 09 decision 5)."""
         parts = []
         for message in waiting:
-            parts.append(conversations.message_text(message))
+            text = conversations.message_text(message)
+            parts.append(f"Differently: {text}" if marked and message["body"].get("tap") == "differently" else text)
             names = message["body"].get("attachment_names", [])
             for ref, name in zip(message["refs"], names):
                 preview = self.runtime.artifacts.read(ref).decode("utf-8", "replace")
@@ -481,6 +498,17 @@ def _act(self, found, own, waiting, said, decision, theta, data_class) -> dict:
             if kept:
                 result["memory"] = kept
                 result["buttons"] = [*result["buttons"], *(f"forget:{m}" for m in kept)][:12]
+        lesson = decision.get("lesson")
+        corrected = any(m["body"].get("tap") == "differently" and m["body"].get("tap_text") for m in waiting)
+        cap = self._channel_cap(channel)
+        if (lesson and corrected and self.store is not None and not result.get("read_files")
+                and memory.looks_secret(lesson["text"]) is None
+                and raised_class(data_class, lesson["text"]) == data_class  # a lesson never raises its decision
+                and (cap is None or ORDER.index(data_class) <= ORDER.index(cap))):
+            result["lesson"] = lesson  # offered only: stored by the remember tap (plan 09 decision 5)
+            result["reply"] = "\n\n".join([result["reply"], say(said, f"🧠 Zapamatovat si: {lesson['text']}?",
+                                                                 f"🧠 Remember: {lesson['text']}?")])
+            result["buttons"] = [*result["buttons"], "remember"][:12]
         return result
 
     def _read_files(self, found, said, context, base, final, called, theta) -> dict: