From 32359c1fe7c6778030ae7c86fa23ccf096dfd56c Mon Sep 17 00:00:00 2001 From: 0xrsydn Date: Wed, 23 Sep 2026 14:17:38 +0700 Subject: [PATCH] feat(policy): add opt-in composite reward scoring with audit tests --- policy/context.py | 1 + policy/reward_scoring.py | 127 +++++++++++++++++++++++++++++++++++++++ policy/selection.py | 4 ++ run.py | 3 +- test_brain.py | 99 ++++++++++++++++++++++++++++++ 5 files changed, 233 insertions(+), 1 deletion(-) create mode 100644 policy/reward_scoring.py diff --git a/policy/context.py b/policy/context.py index 031a187..7ffe9aa 100644 --- a/policy/context.py +++ b/policy/context.py @@ -12,6 +12,7 @@ class Decision: reason: str = "" source: str = "code" # code | jev | fallback confidence: float | None = None + scoring: dict | None = None def __str__(self) -> str: params = ", ".join(f"{k}={v!r}" for k, v in self.params.items()) diff --git a/policy/reward_scoring.py b/policy/reward_scoring.py new file mode 100644 index 0000000..cf53935 --- /dev/null +++ b/policy/reward_scoring.py @@ -0,0 +1,127 @@ +"""Experimental reward-only scoring. No game calls or persistent memory.""" +from __future__ import annotations + +import math + +import facts as F +from jev import ScoreAnswer, gate, score +from .context import Decision +from .scoring import normalize_score, rank_candidates, weighted_utility + +# Fixed experimental weights and margin, not fitted win probabilities. +WEIGHTS = {"offense": 0.35, "defense": 0.30, "scaling": 0.15, "consistency": 0.20} +TAKE_MARGIN = 0.10 +RUBRICS = { + "offense": [ + "Frequently delays attacks or displaces damage needed in the first shuffle.", + "Slightly reduces damage delivered in the first shuffle.", + "Does not materially change damage delivered in the first shuffle.", + "Adds usable damage in the first shuffle with support already observed.", + "Fills a major early damage gap reliably with support already observed.", + ], + "defense": [ + "Frequently prevents playing needed block or direct damage prevention.", + "Slightly reduces available block or direct damage prevention.", + "Does not materially change block or direct damage prevention.", + "Adds usable block or direct damage prevention with existing support.", + "Reliably fills a major block or direct damage prevention gap.", + ], + "scaling": [ + "Frequently disrupts existing repeated-turn damage or defense growth.", + "Slightly delays existing repeated-turn damage or defense growth.", + "Does not materially change performance as a fight continues.", + "Adds usable repeated-turn growth with support already observed.", + "Reliably fills a major long-fight growth gap with existing support.", + ], + "consistency": [ + "Frequently creates unplayable hands or requires missing setup and resources.", + "Adds draw dilution or resource pressure more often than useful access.", + "Does not materially change draw access, resource fit, or setup reliability.", + "Improves draw access, resource fit, or setup using existing support.", + "Reliably removes a major draw, resource, or setup bottleneck.", + ], +} + + +def reward_context(obs: dict, deck: dict | None) -> dict: + """Keep observed metadata intact; do not infer a permanent deck or draw order.""" + player = obs.get("player") or {} + reward = obs.get("card_reward") or {} + return { + "run": obs.get("run") or {}, + "player": {k: player.get(k) for k in + ("character", "hp", "max_hp", "gold", "relics", "potions")}, + "card_evidence": F.deck_context(deck), + "offered": {f"card{c['index']}": dict(c) for c in reward.get("cards", [])}, + "can_skip": reward.get("can_skip"), + "limits": ["Permanent deck is unavailable; combat piles are limited evidence.", + "Generated cards and temporary upgrades may occur in that evidence.", + "Pile order is not draw order. Unobserved support is unknown.", + "Skip side effects and future encounters are unknown."], + } + + +def composite_reward_decision(obs, client, deck, fallback_card) -> Decision: + state = reward_context(obs, deck) + audit = {"policy": "composite-v1", "weights": dict(WEIGHTS), + "take_margin": TAKE_MARGIN, "components": {}, + "evidence": state["card_evidence"], + "skip_baseline": "unknown"} + + def fallback(reason, selected=None): + audit["gate_reason"] = reason + selected = fallback_card if selected is None else selected + return Decision("select_card_reward", {"card_index": selected["index"]}, + f"composite fallback: {reason}; selected {selected['index']}", "fallback", + scoring=audit) + + if client is None: + return fallback("no model client") + if state["card_evidence"] == "unknown": + return fallback("missing scoped card evidence") + + questions = {} + for key in state["offered"]: + for axis, rubric in RUBRICS.items(): + questions[f"{key}_{axis}"] = score( + f"Estimate the marginal {axis} effect of adding offered.{key} versus " + "leaving the deck unchanged. Use only observed support. Consider cost, " + "effect text, duplicates, character mechanics, and relic interactions. " + "Rarity alone is not value. Do not assume future upgrades or pickups. " + "Card pairings matter only when their required support is observed. " + "Do not count faster kills as defense. Do not count synergy as a separate bonus. " + "Use the supplied axis rubric; confidence is not correctness.", rubric) + response = client.ask(state, questions) + usable = True + for key in state["offered"]: + components = {} + for axis in WEIGHTS: + answer = response.get(f"{key}_{axis}") + valid = (isinstance(answer, ScoreAnswer) + and math.isfinite(answer.score) and 0 <= answer.score <= 4 + and len(answer.legend) == 5 and gate(answer)) + components[axis] = {"score": answer.score if isinstance(answer, ScoreAnswer) and math.isfinite(answer.score) else None, + "accepted": bool(valid)} + usable = usable and valid + audit["components"][key] = components + if not usable: + return fallback("missing, invalid, or uncertain component") + + utilities = { + key: weighted_utility( + {axis: normalize_score(value["score"], 0, 2, 4) + for axis, value in components.items()}, + WEIGHTS, + ) + for key, components in audit["components"].items() + } + audit["utilities"] = utilities + best = rank_candidates(utilities)[0] + # The current serializer cannot prove that an alternative is a plain skip. + # Do not silently assign that alternative zero utility. + if utilities[best] <= TAKE_MARGIN and state["can_skip"] is not False: + return fallback("no clear addition; skip effects unresolved", state["offered"][best]) + audit["gate_reason"] = "positive addition" if utilities[best] > TAKE_MARGIN else "forced selection" + return Decision("select_card_reward", {"card_index": state["offered"][best]["index"]}, + f"composite chose {best}; utility={utilities[best]:.3f}", "jev", + scoring=audit) diff --git a/policy/selection.py b/policy/selection.py index e599fa6..7027713 100644 --- a/policy/selection.py +++ b/policy/selection.py @@ -52,6 +52,10 @@ def card_reward_decision(obs: dict, client: JevClient | None, deck: dict | None, if not cards: return Decision("skip_card_reward", {}, "no cards offered", "code") + if skip_policy == "composite": + from .reward_scoring import composite_reward_decision + return composite_reward_decision(obs, client, deck, best_by_rarity(cards)) + if client is None: fallback_card = best_by_rarity(cards) return Decision("select_card_reward", {"card_index": fallback_card["index"]}, diff --git a/run.py b/run.py index 6312ec3..898482e 100644 --- a/run.py +++ b/run.py @@ -128,6 +128,7 @@ def decision_record(step: int, state_type: str, run_state, decision, "params": decision.params, "reason": decision.reason, "confidence": decision.confidence, + "scoring": decision.scoring, "error": error, "jev": getattr(client, "last", None), } @@ -211,7 +212,7 @@ def main() -> int: "session can contain several runs and the results " "cannot be attributed to an arm.") ap.add_argument("--capture-dir", default="capture") - ap.add_argument("--card-skip-policy", choices=("jev", "combined"), default=None, + ap.add_argument("--card-skip-policy", choices=("jev", "combined", "composite"), default=None, help="how card-reward skips are decided (default: brain's own)") args = ap.parse_args() if args.steps < 1 or args.max_jev_errors < 1 or args.max_duplicate_waits < 0: diff --git a/test_brain.py b/test_brain.py index 09672dd..106c7fa 100644 --- a/test_brain.py +++ b/test_brain.py @@ -912,5 +912,104 @@ covered = combat(hand=[card("Defend", index=5, desc="Gain 5 Block.", ctype="Skil check("fully blocked incoming does not force Defend", brain.decide(covered, None).params.get("card_index"), 8) print() +# Composite reward scoring: exercise dispatch with a model stub, never the API. +from policy.selection import card_reward_decision +from policy.reward_scoring import WEIGHTS +from jev import ScoreAnswer + +class CompositeStub: + def __init__(self, values, confidence=0.9): + self.values = values + self.confidence = confidence + self.calls = 0 + + def ask(self, state, questions): + self.calls += 1 + self.state, self.questions = state, questions + return {key: ScoreAnswer(self.values.get(key.split('_')[0], 2), + legend={str(i): text for i, text in enumerate(q['criteria'])}, + confidence=self.confidence) + for key, q in questions.items()} + +reward_obs = {'state_type': 'card_reward', + 'player': {'character': 'Ironclad', 'hp': 20, 'max_hp': 80, 'relics': []}, + 'card_reward': {'can_skip': True, + 'cards': [card('A', 5), card('B', 9, rarity='Rare')]}} +reward_deck = {'counts': {'Strike': 5}, + 'provenance': {'run_id': 'test-run', 'source': 'combat_piles', + 'persistent_deck': False}} +def composite(stub, deck=reward_deck, obs=reward_obs): + return card_reward_decision(obs, stub, deck, skip_policy='composite') + +stub = CompositeStub({'card5': 3.25, 'card9': 2.5}) +result = composite(stub) +check('composite preserves card indices', result.params, {'card_index': 5}) +check('composite batches all axes', len(stub.questions), 8) +check('composite one call', stub.calls, 1) +check('composite preserves fractional score', result.scoring['utilities']['card5'], 0.625) +check('composite logs weights', result.scoring['weights'], WEIGHTS) +check('composite utility is not confidence', result.confidence, None) +check('composite preserves rarity', stub.state['offered']['card9']['rarity'], 'Rare') +check('composite preserves cost', stub.state['offered']['card5']['cost'], reward_obs['card_reward']['cards'][0]['cost']) +check('composite preserves provenance', stub.state['card_evidence']['provenance']['persistent_deck'], False) +for value in (0, 2, float('nan'), 5): + result = composite(CompositeStub({'card5': value, 'card9': value})) + check(f'composite limited fallback {value}', result.source, 'fallback') +check('composite low confidence fallback', composite(CompositeStub({}, 0.1)).source, 'fallback') +unknown = CompositeStub({}) +check('composite unknown deck fallback', composite(unknown, None).source, 'fallback') +check('composite unknown deck makes no call', unknown.calls, 0) +check('composite offline fallback', composite(None).params, {'card_index': 9}) +forced = {**reward_obs, 'card_reward': {**reward_obs['card_reward'], 'can_skip': False}} +check('composite forced selection ranks negative utilities', + composite(CompositeStub({'card5': 1, 'card9': 0}), obs=forced).params, {'card_index': 5}) + +class MissingCompositeStub: + def ask(self, state, questions): + return {} + +check('composite incomplete response fallback', composite(MissingCompositeStub()).source, 'fallback') +from run import decision_record +logged = decision_record(1, 'card_reward', None, result, None, None, session_id='test') +check('runner retains composite audit', logged['scoring'], result.scoring) + +# Pure scoring stays independent of Jev, acceptance rules, and game actions. +from policy.scoring import normalize_score, weighted_utility, rank_candidates + +check('score normalization preserves endpoints and fractions', + [normalize_score(v, 0, 2, 4) for v in (0, 2, 3.25, 4)], [-1, 0, 0.625, 1]) +check('asymmetric scale keeps neutral at zero', + [normalize_score(v, 0, 1, 5) for v in (0.5, 1, 3)], [-0.5, 0, 0.5]) +check('weighted utility preserves negative contributions', + weighted_utility({'a': -1, 'b': 0.5}, {'a': 0.5, 'b': 0.5}), -0.25) +check('ranking is stable and includes negative candidates', + rank_candidates({'first': 0.2, 'last': -1, 'tied': 0.2}), ['first', 'tied', 'last']) +check('empty ranking is explicit', rank_candidates({}), []) + +invalid_scoring_calls = [ + lambda: normalize_score(float('nan'), 0, 2, 4), + lambda: normalize_score(5, 0, 2, 4), + lambda: normalize_score(2, 0, 0, 4), + lambda: weighted_utility({'a': 0}, {'a': 0.5, 'b': 0.5}), + lambda: weighted_utility({'a': None}, {'a': 1}), + lambda: weighted_utility({'a': 0}, {'a': 2}), + lambda: weighted_utility({'a': 0, 'b': 0}, {'a': -1, 'b': 2}), + lambda: weighted_utility({'a': float('inf')}, {'a': 1}), + lambda: rank_candidates({'a': float('nan')}), +] +for i, call in enumerate(invalid_scoring_calls): + try: + call() + except ValueError: + rejected = True + else: + rejected = False + check(f'scoring rejects invalid input {i}', rejected, True) + +check('composite ties retain offer order', + composite(CompositeStub({'card5': 3, 'card9': 3})).params, {'card_index': 5}) +check('composite weak-offer fallback retains highest utility', + composite(CompositeStub({'card5': 1, 'card9': 0})).params, {'card_index': 5}) + print(f"=== {PASS} passed, {FAIL} failed ===") sys.exit(1 if FAIL else 0)