FarmHouse: decision-model client (Laya/Kev) with shadow and assist modes

farmhouse_decide.py asks a /v1/systemone server which action to take; DECISION_MODE=shadow records agreement with the LLM, assist uses it when DECISION_MIN_CONF sure; off by default. scripts/farmhouse_decision_eval.py replays recorded decisions through it. 17 offline tests.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016Er49soaMyG7Yburq57PeB

scripts/farmhouse_decision_eval.py

@@ -0,0 +1,75 @@
+"""How often does a decision model (Laya / Kev) pick the same action as the LLM did, on game states the app has already recorded?
+
+    DECISION_URL=http://laya:8080 python scripts/farmhouse_decision_eval.py [decisions_dir] [model ...]
+
+Reads the brain's own decision logs (logs/decisions/*.jsonl: the snapshot and the LLM's answer), asks the decision model the same
+question for each, and prints agreement overall, at several confidence levels, per action, the "always the most common action"
+baseline, and the time per answer. The LLM is the teacher here, not the truth: this measures how well a fast model can stand in
+for the slow one, not how well either plays.
+"""
+import collections
+import glob
+import importlib.util
+import json
+import os
+import sys
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+spec = importlib.util.spec_from_file_location("farmhouse_decide", os.path.join(HERE, "..", "src", "commander_ai_lab", "sim", "farmhouse_decide.py"))
+D = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(D)
+
+
+def load(dirpath):
+    rows = []
+    for f in sorted(glob.glob(os.path.join(dirpath, "*.jsonl"))):
+        for line in open(f, encoding="utf-8"):
+            try:
+                r = json.loads(line)
+            except ValueError:
+                continue
+            if (r.get("decision") or {}).get("source") == "llm" and r.get("game_state") and r["decision"].get("action") in D.ACTION_HELP:
+                rows.append(r)
+    return rows
+
+
+def main():
+    dirpath = sys.argv[1] if len(sys.argv) > 1 else os.path.join(HERE, "..", "logs", "decisions")
+    models = sys.argv[2:] or [os.environ.get("DECISION_MODEL", "")]
+    rows = load(dirpath)
+    if not rows:
+        print("no recorded LLM decisions in", dirpath)
+        return 1
+    truth = [r["decision"]["action"] for r in rows]
+    common, n_common = collections.Counter(truth).most_common(1)[0]
+    print("%d recorded decisions; LLM action mix: %s" % (len(rows), dict(collections.Counter(truth))))
+    print("baseline 'always %s': %.1f%%\n" % (common, 100.0 * n_common / len(rows)))
+    for model in models:
+        os.environ["DECISION_MODEL"] = model
+        D.STATS.update({"asked": 0, "agreed": 0, "answered_conf": 0, "errors": 0, "ms_total": 0.0, "by_action": {}})
+        got = []
+        for r in rows:
+            a = D.ask(r["game_state"])
+            got.append(a)
+            D.record(a, r["decision"]["action"])
+        ok = [(g, t) for g, t in zip(got, truth) if g]
+        print("== model %r: answered %d of %d, %.0f ms each" % (model or "auto", len(ok), len(rows), D.stats()["avg_ms"] or 0))
+        if not ok:
+            continue
+        print("   agrees with the LLM on %.1f%%  (baseline %.1f%%)" % (100.0 * sum(g["choice"] == t for g, t in ok) / len(ok), 100.0 * n_common / len(rows)))
+        for lo in (0.5, 0.7, 0.85):
+            sel = [(g, t) for g, t in ok if g["confidence"] >= lo]
+            if sel:
+                print("   when it is >= %.2f sure (%d answers, %.0f%%): agrees %.1f%%" % (
+                    lo, len(sel), 100.0 * len(sel) / len(ok), 100.0 * sum(g["choice"] == t for g, t in sel) / len(sel)))
+        by = collections.defaultdict(lambda: [0, 0])
+        for g, t in ok:
+            by[t][0] += 1
+            by[t][1] += g["choice"] == t
+        print("   per LLM action:", {k: "%d/%d" % (v[1], v[0]) for k, v in sorted(by.items())})
+        print("   it picked:", dict(collections.Counter(g["choice"] for g, t in ok)))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/commander_ai_lab/sim/deepseek_brain.py

@@ -254,6 +254,7 @@ def build_game_state_snapshot(
         "my_life": me.life,
         "opp_life": opp.life,
         "my_mana_total": mana_total,
+        "land_play_available": getattr(me, "_lands_played_this_turn", 0) < getattr(me, "_land_plays_this_turn_limit", 1),
         "my_mana_by_color": mana_colors,
         "my_hand": my_hand,
         "my_board": my_board,
@@ -268,6 +269,8 @@ def build_game_state_snapshot(
         "opp_graveyard": opp_gy,
         "my_library_count": cards_remaining,
         "opp_library_count": len(opp.library),
+        "opp_hand_count": len(opp.hand),
+        "stack_supported": False,
         "deck_drawn_pct": pct_drawn,
     }
 
@@ -578,6 +581,11 @@ class DeepSeekBrain:
         if available_mana is not None:
             snapshot["my_mana_available"] = available_mana
 
+        from commander_ai_lab.sim.strategic_context import turn_context
+        advice = turn_context(snapshot, (deck_context or {}).get("strategic_preferences"))
+        snapshot["strategic_advice"] = advice
+        candidate_actions = advice["candidate_actions"]
+
         if self.config.cache_enabled:
             cache_key = self._cache_key(snapshot)
             if cache_key in self._cache:
@@ -587,6 +595,18 @@ class DeepSeekBrain:
                 cached["latency_ms"] = round((time.time() - t_start) * 1000, 1)
                 return cached
 
+        from commander_ai_lab.sim import farmhouse_decide as _fd  # FARMHOUSE_DECISION
+        _fd_mode = _fd.mode()
+        if _fd_mode == "assist":
+            _fd_ans = _fd.ask(snapshot, actions=candidate_actions)
+            if (_fd_ans and _fd_ans["choice"] in candidate_actions
+                    and _fd_ans["confidence"] >= float(os.environ.get("DECISION_MIN_CONF", "0.7"))):
+                result = _fd.as_decision(_fd_ans, t_start)
+                result["strategic_advice"] = advice
+                if self.config.log_decisions:
+                    self._log_decision(snapshot, result, player_index)
+                return result
+
         if not self._connected:
             return self._fallback_action(snapshot, t_start, "not_connected")
 
@@ -595,7 +615,10 @@ class DeepSeekBrain:
             try:
                 result = self._call_llm(snapshot)
                 result["source"] = "llm"
+                if _fd_mode == "shadow":  # FARMHOUSE_DECISION_SHADOW
+                    _fd.record(_fd.ask(snapshot, actions=candidate_actions), result.get("action"))
                 result["latency_ms"] = round((time.time() - t_start) * 1000, 1)
+                result["strategic_advice"] = advice
 
                 if self.config.cache_enabled:
                     self._put_cache(cache_key, result)
@@ -882,4 +905,5 @@ class DeepSeekBrain:
             "api_base": self.config.api_base,
             "pending_log_entries": len(self._decision_log),
             "current_game_id": self._current_game_id,
+            "decision": __import__("commander_ai_lab.sim.farmhouse_decide", fromlist=["stats"]).stats(),  # FARMHOUSE_DECISION_STATS
         }

src/commander_ai_lab/sim/farmhouse_decide.py

@@ -0,0 +1,158 @@
+"""A fast "which action?" answer from a decision model (Laya, or Kev: both serve POST /v1/systemone).
+
+The LLM brain answers the same question about 25 times a game, each taking seconds. A decision model reads a short text and picks ONE of
+a fixed list of options in a single forward pass (tens of milliseconds on a GPU, about a second on a small CPU) and never writes text.
+It cannot choose a target card; the engine's heuristics do that, exactly as when the LLM names no target.
+
+Modes (DECISION_MODE; with no DECISION_URL nothing here runs at all):
+  off     do nothing
+  shadow  ask it too and RECORD whether it agreed with the LLM; the game is played with the LLM's answer (the safe way to measure)
+  assist  ask it first; use its answer when it is at least DECISION_MIN_CONF sure, otherwise ask the LLM as before
+
+Settings: DECISION_URL (e.g. http://laya:8080), DECISION_MODEL (a Laya checkpoint name such as english or typed-decisions),
+DECISION_MIN_CONF (default 0.7, on the probability of the chosen option), DECISION_TIMEOUT (seconds, default 5), DECISION_API_KEY.
+After three failures in a row the model is not asked again for a minute, so a down server costs the game nothing.
+"""
+import json
+import os
+import threading
+import time
+import urllib.request
+
+ACTION_HELP = {
+    "play_land": "play a land from hand (always do this first)",
+    "cast_creature": "cast a creature spell",
+    "cast_removal": "use removal on the opponent's best threat",
+    "cast_board_wipe": "wipe the board when behind on creatures",
+    "cast_ramp": "cast mana ramp to accelerate",
+    "cast_spell": "cast any other spell such as an enchantment or artifact",
+    "attack_all": "attack with all untapped creatures",
+    "attack_safe": "attack only with evasive or large creatures",
+    "hold": "pass and save mana",
+}
+
+_lock = threading.Lock()
+_fail = {"n": 0, "until": 0.0}
+STATS = {"asked": 0, "agreed": 0, "answered_conf": 0, "errors": 0, "ms_total": 0.0, "by_action": {}}
+
+
+def mode():
+    m = os.environ.get("DECISION_MODE", "off").strip().lower()
+    return m if m in ("shadow", "assist") and os.environ.get("DECISION_URL") else "off"
+
+
+def _card_text(c):
+    if isinstance(c, str):
+        return c
+    bits = [str(c.get("type") or "").strip()]
+    if c.get("cmc") is not None:
+        bits.append("cmc %s" % c["cmc"])
+    if c.get("pt"):
+        bits.append(str(c["pt"]))
+    if c.get("castable") is not None:
+        bits.append("castable" if c["castable"] else "not castable")
+    roles = c.get("roles")
+    if roles:
+        bits.append("/".join(roles) if isinstance(roles, (list, tuple)) else str(roles))
+    return "%s (%s)" % (c.get("name", "?"), ", ".join(b for b in bits if b))
+
+
+def compact_state(snap):
+    """The snapshot as a short paragraph (a decision model reads about 500 tokens, the LLM's indented JSON is two to three times that)."""
+    hand = snap.get("my_hand") or []
+    mine = [c for c in (snap.get("my_board") or []) if isinstance(c, dict)]
+    theirs = [c for c in (snap.get("opp_board") or []) if isinstance(c, dict)]
+    my_creatures = [c for c in mine if "creature" in str(c.get("type", "")).lower()]
+    opp_creatures = [c for c in theirs if "creature" in str(c.get("type", "")).lower()]
+    threat = snap.get("opp_threat") or {}
+    cmdr = snap.get("commander") or {}
+    parts = ["Turn %s. Your life %s, opponent life %s. Mana available %s." % (
+        snap.get("turn"), snap.get("my_life"), snap.get("opp_life"), snap.get("my_mana_total", snap.get("my_mana_available")))]
+    parts.append("Opponent threat %s, total power %s." % (threat.get("level", "?"), threat.get("total_power", "?")))
+    if cmdr:
+        parts.append("Your commander %s is in the %s." % (cmdr.get("name", "?"), cmdr.get("zone", "?")))
+    parts.append("Hand: %s." % ("; ".join(_card_text(c) for c in hand[:10]) if hand else "empty"))
+    parts.append("Your creatures: %s." % (", ".join(_card_text(c) for c in my_creatures[:8]) if my_creatures else "none"))
+    parts.append("Your other permanents: %d." % (len(mine) - len(my_creatures)))
+    parts.append("Opponent creatures: %s." % (", ".join(_card_text(c) for c in opp_creatures[:8]) if opp_creatures else "none"))
+    parts.append("Opponent other permanents: %d." % (len(theirs) - len(opp_creatures)))
+    advice = snap.get("strategic_advice") or {}
+    ranking = (advice.get("ranking") or {}).get("ranked") or []
+    if ranking:
+        parts.append("CSIE relative ranking: %s." % "; ".join(
+            "%s %.2f (%s)" % (r["id"], r["score"], r["explanation"])
+            for r in ranking[:5]))
+    if snap.get("stack_supported") and snap.get("stack"):
+        parts.append("Visible stack: %s." % json.dumps(snap["stack"], separators=(",", ":")))
+    return " ".join(parts)
+
+
+def ask(snap, actions=None):
+    """{'choice', 'confidence', 'probabilities', 'ms', 'model'} or None (not configured, backing off, or an error)."""
+    url = os.environ.get("DECISION_URL")
+    if not url:
+        return None
+    now = time.time()
+    with _lock:
+        if _fail["until"] > now:
+            return None
+    acts = [a for a in (actions or ACTION_HELP) if a in ACTION_HELP]
+    body = {"state": {"body": compact_state(snap)},
+            "questions": {"action": {"type": "choice", "instructions": "Which action is best for this turn?",
+                                     "criteria": {a: ACTION_HELP[a] for a in acts}}}}
+    model = os.environ.get("DECISION_MODEL")
+    if model:
+        body["model"] = model
+    headers = {"content-type": "application/json"}
+    if os.environ.get("DECISION_API_KEY"):
+        headers["authorization"] = "Bearer " + os.environ["DECISION_API_KEY"]
+    t0 = time.time()
+    try:
+        req = urllib.request.Request(url.rstrip("/") + "/v1/systemone", data=json.dumps(body).encode(), headers=headers)
+        out = json.load(urllib.request.urlopen(req, timeout=float(os.environ.get("DECISION_TIMEOUT", "5"))))
+        ans = out["answers"]["action"]
+        probs = ans.get("probabilities") or {}
+        choice = ans["choice"]
+        if choice not in acts:
+            raise ValueError("Decision model selected an action outside the supplied candidates")
+        with _lock:
+            _fail["n"] = 0
+        return {"choice": choice, "confidence": float(probs.get(choice, ans.get("answer_confidence", 0.0))),
+                "probabilities": probs, "ms": round((time.time() - t0) * 1000, 1), "model": model or "auto"}
+    except Exception:
+        with _lock:
+            _fail["n"] += 1
+            STATS["errors"] += 1
+            if _fail["n"] >= 3:
+                _fail["until"] = time.time() + 60
+                _fail["n"] = 0
+        return None
+
+
+def as_decision(answer, t_start):
+    """The brain's own result dict for a decision model's answer (no target card: the engine's heuristics pick it)."""
+    return {"action": answer["choice"], "target_card": "", "source": "decision",
+            "reasoning": "decision model %s, p=%.2f" % (answer["model"], answer["confidence"]),
+            "latency_ms": round((time.time() - t_start) * 1000, 1)}
+
+
+def record(answer, llm_action):
+    """Count one shadow comparison."""
+    if not answer:
+        return
+    with _lock:
+        STATS["asked"] += 1
+        STATS["ms_total"] += answer["ms"]
+        agree = answer["choice"] == llm_action
+        STATS["agreed"] += 1 if agree else 0
+        if answer["confidence"] >= float(os.environ.get("DECISION_MIN_CONF", "0.7")):
+            STATS["answered_conf"] += 1
+        row = STATS["by_action"].setdefault(llm_action, {"n": 0, "agreed": 0})
+        row["n"] += 1
+        row["agreed"] += 1 if agree else 0
+
+
+def stats():
+    with _lock:
+        n = STATS["asked"]
+        return dict(STATS, agreement=round(STATS["agreed"] / n, 3) if n else None, avg_ms=round(STATS["ms_total"] / n, 1) if n else None)

tests/farmhouse/test_farmhouse_decide.py

@@ -0,0 +1,111 @@
+"""Run:  python -I -X utf8 tests/farmhouse/test_farmhouse_decide.py   (offline: a stub /v1/systemone server on localhost)"""
+import json
+import os
+import sys
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from _load import load  # noqa: E402
+
+D = load("farmhouse_decide", "src/commander_ai_lab/sim/farmhouse_decide.py")
+FAILS = 0
+
+
+def ck(name, cond, detail=""):
+    global FAILS
+    print("  %s  %s%s" % ("ok  " if cond else "FAIL", name, ("  -- " + str(detail)[:200]) if detail and not cond else ""))
+    if not cond:
+        FAILS += 1
+
+
+SEEN = []
+BEHAVIOUR = {"mode": "ok"}
+
+
+class Stub(BaseHTTPRequestHandler):
+    def log_message(self, *a):
+        pass
+
+    def do_POST(self):
+        body = json.loads(self.rfile.read(int(self.headers.get("Content-Length") or 0)))
+        SEEN.append((self.path, body, self.headers.get("authorization")))
+        if BEHAVIOUR["mode"] == "down":
+            self.send_response(500)
+            self.end_headers()
+            return
+        crit = list(body["questions"]["action"]["criteria"])
+        pick = "cast_creature" if "cast_creature" in crit else crit[0]
+        probs = {k: (0.8 if k == pick else 0.2 / max(len(crit) - 1, 1)) for k in crit}
+        out = json.dumps({"answers": {"action": {"type": "choice", "choice": pick, "probabilities": probs, "confidence": 0.5}}}).encode()
+        self.send_response(200)
+        self.send_header("content-type", "application/json")
+        self.send_header("content-length", str(len(out)))
+        self.end_headers()
+        self.wfile.write(out)
+
+
+srv = HTTPServer(("127.0.0.1", 0), Stub)
+threading.Thread(target=srv.serve_forever, daemon=True).start()
+URL = "http://127.0.0.1:%d" % srv.server_port
+
+SNAP = {"turn": 4, "my_life": 31, "opp_life": 20, "my_mana_total": 4,
+        "my_hand": [{"name": "Forest", "type": "land", "cmc": 0, "castable": True},
+                    {"name": "Llanowar Elves", "type": "creature", "cmc": 1, "castable": True, "roles": ["ramp"]}, "Sol Ring"],
+        "my_board": [{"name": "Bear", "type": "creature", "pt": "2/2"}, {"name": "Rock", "type": "artifact"}],
+        "opp_board": [{"name": "Wurm", "type": "creature", "pt": "6/6"}],
+        "opp_threat": {"level": "HIGH", "total_power": 6}, "commander": {"name": "Atraxa", "zone": "command zone"}}
+
+# --- the text it reads
+t = D.compact_state(SNAP)
+ck("the text has the facts the choice depends on", all(x in t for x in ("Turn 4", "life 31", "Llanowar Elves", "Bear (creature, 2/2)", "Wurm", "HIGH")), t)
+ck("short enough for a 512-token reader", len(D.compact_state(dict(SNAP, my_hand=[{"name": "Card %d" % i, "type": "creature", "cmc": 3} for i in range(40)]))) < 1900)
+ck("an empty snapshot does not raise", isinstance(D.compact_state({}), str))
+
+# --- modes
+for k in ("DECISION_URL", "DECISION_MODE", "DECISION_MODEL", "DECISION_API_KEY"):
+    os.environ.pop(k, None)
+ck("no URL: off, and ask() does nothing", D.mode() == "off" and D.ask(SNAP) is None and not SEEN)
+os.environ["DECISION_URL"] = URL
+ck("a URL alone is still off (the mode must be chosen)", D.mode() == "off")
+os.environ["DECISION_MODE"] = "shadow"
+ck("shadow", D.mode() == "shadow")
+os.environ["DECISION_MODE"] = "assist"
+ck("assist", D.mode() == "assist")
+os.environ["DECISION_MODE"] = "bogus"
+ck("an unknown mode is off", D.mode() == "off")
+os.environ["DECISION_MODE"] = "shadow"
+
+# --- asking
+os.environ["DECISION_MODEL"] = "typed-decisions"
+os.environ["DECISION_API_KEY"] = "k"
+a = D.ask(SNAP, actions=["play_land", "cast_creature", "hold", "not_an_action"])
+path, body, auth = SEEN[-1]
+ck("it posts to /v1/systemone with the model, key and only the known actions",
+   path == "/v1/systemone" and body["model"] == "typed-decisions" and auth == "Bearer k"
+   and list(body["questions"]["action"]["criteria"]) == ["play_land", "cast_creature", "hold"], (path, body, auth))
+ck("the answer carries the choice, its probability and timing", a and a["choice"] == "cast_creature" and abs(a["confidence"] - 0.8) < 1e-9 and a["ms"] >= 0, a)
+d = D.as_decision(a, time.time())
+ck("as_decision is a brain result with no target (the engine picks one)", d["action"] == "cast_creature" and d["target_card"] == "" and d["source"] == "decision", d)
+
+# --- shadow bookkeeping
+D.record(a, "cast_creature")
+D.record(a, "hold")
+D.record(None, "hold")
+s = D.stats()
+ck("agreement is counted over answered questions only", s["asked"] == 2 and s["agreed"] == 1 and s["agreement"] == 0.5 and s["answered_conf"] == 2, s)
+ck("per-action tallies", s["by_action"]["hold"] == {"n": 1, "agreed": 0} and s["by_action"]["cast_creature"]["agreed"] == 1)
+
+# --- a server that is down costs nothing after three failures
+BEHAVIOUR["mode"] = "down"
+n0 = len(SEEN)
+res = [D.ask(SNAP) for _ in range(6)]
+ck("errors give None, never raise", all(r is None for r in res))
+ck("after three failures it stops asking for a while", len(SEEN) - n0 == 3, len(SEEN) - n0)
+BEHAVIOUR["mode"] = "ok"
+ck("still backing off even though the server is back", D.ask(SNAP) is None)
+D._fail["until"] = 0
+ck("and asks again once the pause ends", D.ask(SNAP) is not None)
+print("\nFAILURES: %d" % FAILS)
+sys.exit(1 if FAILS else 0)