serversdown
fc623db27a
Storage + tools so Decide mode can learn instead of one-shot tie-breaking: log the call Brian makes, resolve it later with the outcome, recall similar past calls to ground new recommendations in his own track record. - memory: decisions table + Decision dataclass + log/resolve/get/list/recall_decisions (embedding over situation+choice; embed failure never blocks a log) - tools: log_decision / resolve_decision / recall_decisions handlers + specs - tests: 9 covering roundtrip, resolve, open-only filter, similarity rank, tool layer Data layer only — NOT wired into any mode's allow-list or the Decide card yet (the prompt/taste part is left for Brian; see docs/DECISION_LOG.md). No behavior changes until the tools are added to _DECIDE_TOOLS. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
104 lines
4.0 KiB
Python
104 lines
4.0 KiB
Python
"""Decision log (Decide mode's learning layer): log -> resolve -> recall, + tools."""
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def mem(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("LYRA_DB_PATH", str(tmp_path / "test.db"))
|
|
from lyra import llm
|
|
# Deterministic, content-dependent embeddings so recall ordering is meaningful:
|
|
# "cleveland"/"tournament" cluster on axis 0, "stocks"/"money" on axis 1.
|
|
def fake_embed(texts):
|
|
out = []
|
|
for t in texts:
|
|
t = t.lower()
|
|
poker = sum(w in t for w in ("tournament", "cleveland", "poker", "buy-in"))
|
|
money = sum(w in t for w in ("stocks", "money", "invest", "sell"))
|
|
out.append([float(poker), float(money), 0.1])
|
|
return out
|
|
monkeypatch.setattr(llm, "embed", fake_embed)
|
|
import lyra.memory as memory
|
|
importlib.reload(memory)
|
|
return memory
|
|
|
|
|
|
def test_log_and_get_roundtrip(mem):
|
|
did = mem.log_decision(
|
|
situation="Play the Cleveland turbo tournament tomorrow?",
|
|
choice="Yes, but only the noon flight",
|
|
options="skip it / noon flight / both flights",
|
|
rationale="20-min levels suit my aggression; one flight caps the variance",
|
|
confidence=4, tags="poker,tournament",
|
|
)
|
|
d = mem.get_decision(did)
|
|
assert d.situation.startswith("Play the Cleveland")
|
|
assert d.choice == "Yes, but only the noon flight"
|
|
assert d.confidence == 4 and d.tags == "poker,tournament"
|
|
assert not d.resolved and d.outcome is None
|
|
|
|
|
|
def test_resolve_closes_the_loop(mem):
|
|
did = mem.log_decision(situation="Sell the stocks now?", choice="Hold")
|
|
assert mem.resolve_decision(did, "Recovered 12% the next week", outcome_rating=1)
|
|
d = mem.get_decision(did)
|
|
assert d.resolved and d.outcome_rating == 1
|
|
assert "Recovered" in d.outcome and d.resolved_at is not None
|
|
|
|
|
|
def test_resolve_unknown_id_is_false(mem):
|
|
assert mem.resolve_decision(999, "n/a") is False
|
|
|
|
|
|
def test_list_open_only_filters_resolved(mem):
|
|
a = mem.log_decision(situation="A?", choice="x")
|
|
mem.log_decision(situation="B?", choice="y")
|
|
mem.resolve_decision(a, "done", 0)
|
|
assert {d.situation for d in mem.list_decisions(open_only=True)} == {"B?"}
|
|
assert len(mem.list_decisions()) == 2
|
|
|
|
|
|
def test_recall_ranks_by_similarity(mem):
|
|
mem.log_decision(situation="Which Cleveland tournament flight?", choice="noon")
|
|
mem.log_decision(situation="Should I sell the stocks?", choice="hold")
|
|
hits = mem.recall_decisions("another poker tournament buy-in", k=2)
|
|
assert hits[0].situation.startswith("Which Cleveland") # poker cluster ranks first
|
|
assert hits[0].score >= hits[1].score
|
|
|
|
|
|
# --- tool layer ---------------------------------------------------------------
|
|
|
|
def test_log_decision_tool_persists(mem):
|
|
from lyra import tools
|
|
out = tools.dispatch("log_decision",
|
|
{"situation": "Move the MI50 to auto clocks?", "choice": "yes",
|
|
"confidence": "3", "tags": "build"})
|
|
assert "#1" in out
|
|
d = mem.get_decision(1)
|
|
assert d.choice == "yes" and d.confidence == 3 and d.tags == "build"
|
|
|
|
|
|
def test_log_decision_tool_requires_both_fields(mem):
|
|
from lyra import tools
|
|
assert "Need both" in tools.dispatch("log_decision", {"situation": "just this"})
|
|
|
|
|
|
def test_resolve_decision_tool(mem):
|
|
from lyra import tools
|
|
did = mem.log_decision(situation="X?", choice="y")
|
|
out = tools.dispatch("resolve_decision",
|
|
{"decision_id": did, "outcome": "worked out", "rating": "1"})
|
|
assert f"#{did}" in out
|
|
assert mem.get_decision(did).outcome_rating == 1
|
|
|
|
|
|
def test_recall_decisions_tool_surfaces_outcomes(mem):
|
|
from lyra import tools
|
|
did = mem.log_decision(situation="Cleveland tournament again?", choice="play")
|
|
mem.resolve_decision(did, "min-cashed", outcome_rating=0)
|
|
out = tools.dispatch("recall_decisions", {"query": "poker tournament tomorrow"})
|
|
assert "Cleveland" in out and "mixed" in out
|