feat(policy): add opt-in composite reward scoring with audit tests

This commit is contained in:
0xrsydn 2026-09-23 14:17:38 +07:00
commit 32359c1fe7
5 changed files with 233 additions and 1 deletions

View file

@ -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)