wehub-resource-sync
628 lines
22 KiB
Python
628 lines
22 KiB
Python
"""Explicit-feedback consumption: alignment reasons, weights, gates, rollback.
|
|
|
|
The load-bearing invariant: ``align_session(samples, feedback=None)`` must be
|
|
byte-identical to the pre-feedback behavior — pinned first, below.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from opensquilla.squilla_router.self_learning.alignment import (
|
|
REASON_CONFIDENCE_BACKOFF,
|
|
REASON_EXPLICIT_DOWNVOTE,
|
|
REASON_EXPLICIT_DOWNVOTE_ENSEMBLE,
|
|
REASON_EXPLICIT_DOWNVOTE_HIGH_TIER,
|
|
REASON_EXPLICIT_UPVOTE,
|
|
REASON_EXPLICIT_UPVOTE_ENSEMBLE,
|
|
REASON_IMMEDIATE_COMPLAINT,
|
|
REASON_NORMAL,
|
|
REASON_RETROSPECTIVE,
|
|
align_session,
|
|
)
|
|
from opensquilla.squilla_router.self_learning.feedback import FeedbackEntry
|
|
from opensquilla.squilla_router.self_learning.schema import (
|
|
RouterTrainSample,
|
|
encode_features,
|
|
)
|
|
|
|
SESSION = "agent:main:webchat:s1"
|
|
|
|
|
|
def _sample(
|
|
turn: int,
|
|
*,
|
|
route: str = "R0",
|
|
final: str | None = None,
|
|
complaint: bool = False,
|
|
gate: bool = False,
|
|
seed: float = 0.0,
|
|
) -> RouterTrainSample:
|
|
return RouterTrainSample(
|
|
session_key=SESSION,
|
|
turn_index=turn,
|
|
ts=f"2026-07-01T00:00:{turn:02d}Z",
|
|
feature_schema_version="v1",
|
|
features_390_b64=encode_features(np.full(390, seed, np.float32)),
|
|
route_class=route,
|
|
final_route_class=final or route,
|
|
complaint_detected=complaint,
|
|
confidence_gate_applied=gate,
|
|
decision_id=f"dec-{turn}",
|
|
)
|
|
|
|
|
|
def _fb(rating: str, kind: str = "single") -> FeedbackEntry:
|
|
return FeedbackEntry(rating=rating, executed_kind=kind)
|
|
|
|
|
|
def _fbmap(turn: int, rating: str, kind: str = "single") -> dict:
|
|
return {f"dec-{turn}": _fb(rating, kind)}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Regression anchor: feedback=None keeps the exact pre-feedback output
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_no_feedback_is_byte_identical_to_legacy() -> None:
|
|
samples = [
|
|
_sample(0, route="R0", seed=0.1),
|
|
_sample(1, route="R0", final="R2", complaint=True, seed=0.2),
|
|
_sample(2, route="R1", gate=True, seed=0.3),
|
|
_sample(3, route="R3", seed=0.4),
|
|
]
|
|
default = align_session(samples)
|
|
explicit_none = align_session(samples, None)
|
|
empty = align_session(samples, {})
|
|
|
|
for legacy, a, b in zip(default, explicit_none, empty, strict=True):
|
|
for other in (a, b):
|
|
assert legacy.reason == other.reason
|
|
assert legacy.target_idx == other.target_idx
|
|
assert legacy.confirmed == other.confirmed
|
|
assert np.array_equal(legacy.features_390, other.features_390)
|
|
|
|
# Reasons are exactly the legacy set on this fixture.
|
|
assert [s.reason for s in default] == [
|
|
REASON_RETROSPECTIVE, # turn 0: turn 1 complains, R0 eligible
|
|
REASON_IMMEDIATE_COMPLAINT,
|
|
REASON_CONFIDENCE_BACKOFF,
|
|
REASON_NORMAL,
|
|
]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Down-vote semantics (single-model)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_downvote_on_underrouted_turn_upgrades_one_step() -> None:
|
|
samples = [_sample(0, route="R0"), _sample(1, route="R0")]
|
|
fb = _fbmap(0, "down")
|
|
|
|
aligned = align_session(samples, fb)
|
|
|
|
assert aligned[0].reason == REASON_EXPLICIT_DOWNVOTE
|
|
assert aligned[0].target_idx == 1 # +1 step only, not retro's +2
|
|
assert aligned[0].confirmed is True # calm follow-up confirms
|
|
|
|
|
|
def test_downvote_without_followup_is_unconfirmed() -> None:
|
|
samples = [_sample(0, route="R0")]
|
|
aligned = align_session(samples, _fbmap(0, "down"))
|
|
assert aligned[0].reason == REASON_EXPLICIT_DOWNVOTE
|
|
assert aligned[0].confirmed is False
|
|
|
|
|
|
def test_downvote_on_high_prediction_never_upgrades() -> None:
|
|
"""R2/R3 predictions: dissatisfaction is not attributable to the tier."""
|
|
samples = [_sample(0, route="R2"), _sample(1, route="R3")]
|
|
fb = {**_fbmap(0, "down"), **_fbmap(1, "down")}
|
|
|
|
aligned = align_session(samples, fb)
|
|
|
|
for a, expected_target in zip(aligned, (2, 3), strict=True):
|
|
assert a.reason == REASON_EXPLICIT_DOWNVOTE_HIGH_TIER # excluded bucket
|
|
assert a.target_idx == expected_target # target unchanged — no upgrade
|
|
|
|
|
|
def test_downvote_endorses_existing_complaint_upgrade() -> None:
|
|
samples = [_sample(0, route="R0", final="R2", complaint=True)]
|
|
aligned = align_session(samples, _fbmap(0, "down"))
|
|
|
|
assert aligned[0].reason == REASON_EXPLICIT_DOWNVOTE # upgraded from complaint
|
|
assert aligned[0].target_idx == 2 # complaint target kept
|
|
|
|
|
|
def test_downvote_on_capped_complaint_does_not_endorse_served_tier() -> None:
|
|
"""A complaint whose upgrade was capped (final == route) is not a real
|
|
upgrade; the down-vote must take the standalone +1 path, never train the
|
|
rejected tier at the table's highest weight."""
|
|
samples = [_sample(0, route="R0", final="R0", complaint=True)]
|
|
aligned = align_session(samples, _fbmap(0, "down"))
|
|
|
|
assert aligned[0].reason == REASON_EXPLICIT_DOWNVOTE
|
|
assert aligned[0].target_idx == 1 # +1 from served, NOT the served R0
|
|
|
|
|
|
def test_downvote_endorses_retrospective() -> None:
|
|
samples = [
|
|
_sample(0, route="R0"),
|
|
_sample(1, route="R0", final="R2", complaint=True),
|
|
_sample(2, route="R0"),
|
|
]
|
|
aligned = align_session(samples, _fbmap(0, "down"))
|
|
|
|
assert aligned[0].reason == REASON_EXPLICIT_DOWNVOTE
|
|
assert aligned[0].target_idx == 2 # retro target (resolving tier) kept
|
|
|
|
|
|
def test_next_turn_complaint_keeps_downvote_unconfirmed() -> None:
|
|
samples = [
|
|
_sample(0, route="R0"),
|
|
_sample(1, route="R0", final="R1", complaint=True),
|
|
_sample(2, route="R0"),
|
|
]
|
|
# Turn 0 down-voted AND turn 1 complains: retro fires first (target R1),
|
|
# the down-vote endorses it.
|
|
aligned = align_session(samples, _fbmap(0, "down"))
|
|
assert aligned[0].reason == REASON_EXPLICIT_DOWNVOTE
|
|
assert aligned[0].target_idx == 1
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Ensemble split
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_ensemble_downvote_never_upgrades() -> None:
|
|
samples = [_sample(0, route="R0")]
|
|
aligned = align_session(samples, _fbmap(0, "down", "ensemble"))
|
|
|
|
assert aligned[0].reason == REASON_EXPLICIT_DOWNVOTE_ENSEMBLE
|
|
assert aligned[0].target_idx == 0 # no upgrade even though R0-eligible
|
|
|
|
|
|
def test_ensemble_upvote_uses_diluted_reason() -> None:
|
|
samples = [_sample(0, route="R1")]
|
|
aligned = align_session(samples, _fbmap(0, "up", "ensemble"))
|
|
assert aligned[0].reason == REASON_EXPLICIT_UPVOTE_ENSEMBLE
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Up-vote semantics
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_upvote_confirms_served_tier() -> None:
|
|
samples = [_sample(0, route="R1")]
|
|
aligned = align_session(samples, _fbmap(0, "up"))
|
|
assert aligned[0].reason == REASON_EXPLICIT_UPVOTE
|
|
assert aligned[0].target_idx == 1
|
|
|
|
|
|
def test_upvote_never_overrides_corrections() -> None:
|
|
"""An up-vote must not weaken complaint/retro corrections on the turn."""
|
|
complaint = [_sample(0, route="R0", final="R2", complaint=True)]
|
|
aligned = align_session(complaint, _fbmap(0, "up"))
|
|
assert aligned[0].reason == REASON_IMMEDIATE_COMPLAINT
|
|
|
|
retro = [
|
|
_sample(0, route="R0"),
|
|
_sample(1, route="R0", final="R2", complaint=True),
|
|
_sample(2, route="R0"),
|
|
]
|
|
aligned2 = align_session(retro, _fbmap(0, "up"))
|
|
assert aligned2[0].reason == REASON_RETROSPECTIVE
|
|
|
|
|
|
def test_upvote_overrides_backoff() -> None:
|
|
samples = [_sample(0, route="R1", gate=True)]
|
|
aligned = align_session(samples, _fbmap(0, "up"))
|
|
assert aligned[0].reason == REASON_EXPLICIT_UPVOTE
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Dataset weights
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def _weights_for(aligned_reasons_and_flags):
|
|
"""Build minimal AlignedSample list and run _compute_weights."""
|
|
from opensquilla.squilla_router.self_learning.alignment import AlignedSample
|
|
from opensquilla.squilla_router.self_learning.dataset import _compute_weights
|
|
|
|
aligned = []
|
|
for i, (reason, confirmed) in enumerate(aligned_reasons_and_flags):
|
|
aligned.append(
|
|
AlignedSample(
|
|
features_390=np.full(390, float(i), np.float32),
|
|
target_idx=1,
|
|
served_idx=1,
|
|
reason=reason,
|
|
session_key=SESSION,
|
|
turn_index=i,
|
|
day="2026-07-01",
|
|
feature_hash=f"h{i}", # distinct → no flood damping
|
|
confirmed=confirmed,
|
|
)
|
|
)
|
|
return _compute_weights(aligned)
|
|
|
|
|
|
def test_weight_table_and_exclusion() -> None:
|
|
weights = _weights_for(
|
|
[
|
|
(REASON_EXPLICIT_DOWNVOTE, True),
|
|
(REASON_RETROSPECTIVE, True),
|
|
(REASON_EXPLICIT_UPVOTE, True),
|
|
(REASON_EXPLICIT_UPVOTE_ENSEMBLE, True),
|
|
(REASON_EXPLICIT_DOWNVOTE_ENSEMBLE, True),
|
|
(REASON_EXPLICIT_DOWNVOTE_HIGH_TIER, True),
|
|
(REASON_NORMAL, True),
|
|
]
|
|
)
|
|
assert weights[0] == pytest.approx(1.2) # downvote > retro
|
|
assert weights[1] == pytest.approx(1.0)
|
|
assert weights[2] == pytest.approx(0.6)
|
|
assert weights[3] == pytest.approx(0.3) # ensemble upvote = normal level
|
|
assert weights[4] == 0.0 # excluded entirely
|
|
assert weights[5] == 0.0 # high-tier exclusion, distinct reason
|
|
assert weights[6] == pytest.approx(0.3)
|
|
|
|
|
|
def test_unconfirmed_downvote_halved() -> None:
|
|
weights = _weights_for([(REASON_EXPLICIT_DOWNVOTE, False)])
|
|
assert weights[0] == pytest.approx(0.6) # 1.2 * 0.5
|
|
|
|
|
|
def test_upvote_flood_damping() -> None:
|
|
"""Identical feature vectors: repeated up-votes are damped like normals."""
|
|
from opensquilla.squilla_router.self_learning.alignment import AlignedSample
|
|
from opensquilla.squilla_router.self_learning.dataset import _compute_weights
|
|
|
|
aligned = [
|
|
AlignedSample(
|
|
features_390=np.zeros(390, np.float32),
|
|
target_idx=1,
|
|
served_idx=1,
|
|
reason=REASON_EXPLICIT_UPVOTE,
|
|
session_key=SESSION,
|
|
turn_index=i,
|
|
day="2026-07-01",
|
|
feature_hash="same", # identical vector recurring
|
|
confirmed=True,
|
|
)
|
|
for i in range(4)
|
|
]
|
|
weights = _compute_weights(aligned)
|
|
assert weights[0] == pytest.approx(0.6 / 2.0) # 0.6 / sqrt(4)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Dataset join end-to-end (store -> feedback -> aligned matrix)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_build_dataset_joins_feedback(tmp_path) -> None:
|
|
from opensquilla.squilla_router.self_learning.dataset import build_training_dataset
|
|
from opensquilla.squilla_router.self_learning.feedback import write_feedback
|
|
from opensquilla.squilla_router.self_learning.store import write_sample
|
|
|
|
for turn in range(3):
|
|
write_sample(_sample(turn, route="R0", seed=float(turn)), "main", home=tmp_path)
|
|
write_feedback(
|
|
"main",
|
|
decision_id="dec-0",
|
|
session_key=SESSION,
|
|
turn_index=0,
|
|
rating="down",
|
|
home=tmp_path,
|
|
)
|
|
|
|
ds = build_training_dataset("main", home=tmp_path)
|
|
|
|
reasons = dict(zip(ds.turn_indices, ds.reasons, strict=True))
|
|
assert reasons[0] == REASON_EXPLICIT_DOWNVOTE
|
|
assert reasons[1] == REASON_NORMAL
|
|
labels = dict(zip(ds.turn_indices, ds.y.tolist(), strict=True))
|
|
assert labels[0] == 1 # R0 -> R1 upgrade label
|
|
|
|
|
|
def test_build_dataset_without_feedback_unchanged(tmp_path) -> None:
|
|
"""No sidecar present: dataset identical to the pre-feedback pipeline."""
|
|
from opensquilla.squilla_router.self_learning.dataset import build_training_dataset
|
|
from opensquilla.squilla_router.self_learning.store import write_sample
|
|
|
|
for turn in range(3):
|
|
write_sample(_sample(turn, route="R0", seed=float(turn)), "main", home=tmp_path)
|
|
|
|
ds = build_training_dataset("main", home=tmp_path)
|
|
assert set(ds.reasons) == {REASON_NORMAL}
|
|
assert ds.y.tolist() == [0, 0, 0]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Rollback second trigger
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_downvote_rate_triggers_rollback() -> None:
|
|
from types import SimpleNamespace
|
|
|
|
from opensquilla.squilla_router.self_learning.promotion import should_rollback
|
|
|
|
cfg = SimpleNamespace()
|
|
# Complaint rate flat; downvote rate jumped 0.0 -> 0.4 on 6 ratings.
|
|
assert should_rollback(
|
|
pre_complaint_rate=0.1,
|
|
post_complaint_rate=0.1,
|
|
post_n=100,
|
|
config=cfg,
|
|
pre_downvote_rate=0.0,
|
|
post_downvote_rate=0.4,
|
|
post_feedback_n=6,
|
|
)
|
|
|
|
|
|
def test_downvote_trigger_respects_min_samples() -> None:
|
|
from types import SimpleNamespace
|
|
|
|
from opensquilla.squilla_router.self_learning.promotion import should_rollback
|
|
|
|
assert not should_rollback(
|
|
pre_complaint_rate=0.1,
|
|
post_complaint_rate=0.1,
|
|
post_n=100,
|
|
config=SimpleNamespace(),
|
|
pre_downvote_rate=0.0,
|
|
post_downvote_rate=1.0,
|
|
post_feedback_n=4, # below min_feedback_monitor_samples default 5
|
|
)
|
|
|
|
|
|
def test_complaint_trigger_unchanged() -> None:
|
|
from types import SimpleNamespace
|
|
|
|
from opensquilla.squilla_router.self_learning.promotion import should_rollback
|
|
|
|
assert should_rollback(
|
|
pre_complaint_rate=0.1,
|
|
post_complaint_rate=0.3,
|
|
post_n=30,
|
|
config=SimpleNamespace(),
|
|
)
|
|
assert not should_rollback(
|
|
pre_complaint_rate=0.1,
|
|
post_complaint_rate=0.12,
|
|
post_n=30,
|
|
config=SimpleNamespace(),
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Volume gate counts down-votes
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_gate_counts_feedback_down_toward_volume() -> None:
|
|
from opensquilla.gateway.config import RouterSelfLearningConfig
|
|
from opensquilla.squilla_router.self_learning.gates import evaluate_training_gates
|
|
from opensquilla.squilla_router.self_learning.state import EventStoreStats, TrainState
|
|
|
|
cfg = RouterSelfLearningConfig(
|
|
enabled=True, train_min_samples=10, idle_hours=0.0, cooldown_hours=0.0
|
|
)
|
|
stats = EventStoreStats(
|
|
total=50, high_value=6, distinct_classes=2, last_ts="2026-01-01T00:00:00Z"
|
|
)
|
|
gate = evaluate_training_gates(config=cfg, state=TrainState(), stats=stats)
|
|
assert not gate.should_train # 6 < 10
|
|
|
|
stats_fb = EventStoreStats(
|
|
total=50,
|
|
high_value=6,
|
|
distinct_classes=2,
|
|
last_ts="2026-01-01T00:00:00Z",
|
|
feedback_down=4,
|
|
)
|
|
gate2 = evaluate_training_gates(config=cfg, state=TrainState(), stats=stats_fb)
|
|
assert gate2.should_train # 6 + 4 >= 10
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Orchestrator end-to-end with feedback (promote baseline / rollback / receipt)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def _write_min_store(tmp_path, n_high=8) -> None:
|
|
from opensquilla.squilla_router.self_learning.store import write_sample
|
|
|
|
for i in range(n_high):
|
|
s = RouterTrainSample(
|
|
session_key="agent:main:webchat:s1",
|
|
turn_index=i,
|
|
ts=f"2026-06-01T00:00:{i:02d}Z",
|
|
feature_schema_version="v1",
|
|
features_390_b64=encode_features(np.full(390, float(i), np.float32)),
|
|
route_class="R0" if i % 2 else "R1",
|
|
final_route_class="R2" if i % 2 else "R3",
|
|
complaint_detected=True,
|
|
)
|
|
write_sample(s, "main", home=tmp_path)
|
|
calm = RouterTrainSample(
|
|
session_key="agent:main:webchat:s2",
|
|
turn_index=0,
|
|
ts="2026-06-01T01:00:00Z",
|
|
feature_schema_version="v1",
|
|
features_390_b64=encode_features(np.zeros(390, np.float32)),
|
|
route_class="R1",
|
|
final_route_class="R1",
|
|
)
|
|
write_sample(calm, "main", home=tmp_path)
|
|
|
|
|
|
def test_orchestrator_records_and_clears_downvote_baseline(tmp_path) -> None:
|
|
from datetime import UTC, datetime
|
|
|
|
from opensquilla.gateway.config import RouterSelfLearningConfig, SquillaRouterConfig
|
|
from opensquilla.squilla_router.self_learning.feedback import write_feedback
|
|
from opensquilla.squilla_router.self_learning.orchestrator import (
|
|
in_process_trainer,
|
|
maybe_run_update_router,
|
|
)
|
|
from opensquilla.squilla_router.self_learning.state import load_train_state
|
|
|
|
now = datetime(2026, 6, 6, 12, 0, 0, tzinfo=UTC)
|
|
_write_min_store(tmp_path)
|
|
# Pre-promotion feedback: 1 down / 4 up (single) -> baseline 0.2.
|
|
for i, rating in enumerate(["down", "up", "up", "up", "up"]):
|
|
write_feedback(
|
|
"main",
|
|
decision_id=f"pre-{i}",
|
|
session_key="agent:main:webchat:s1",
|
|
turn_index=i,
|
|
rating=rating,
|
|
home=tmp_path,
|
|
now=datetime(2026, 6, 2, 0, 0, i, tzinfo=UTC),
|
|
)
|
|
|
|
cfg = SquillaRouterConfig(
|
|
self_learning=RouterSelfLearningConfig(
|
|
enabled=True,
|
|
train_min_samples=4,
|
|
idle_hours=2.0,
|
|
cooldown_hours=72.0,
|
|
holdout_min_size=4,
|
|
holdout_pct=0.4,
|
|
holdout_repeats=2,
|
|
max_critical_under_routing=1.0,
|
|
cost_tolerance_pct=500.0,
|
|
)
|
|
)
|
|
(tmp_path / "base").mkdir() # empty base -> trains fresh
|
|
res = maybe_run_update_router(
|
|
"main",
|
|
router_cfg=cfg,
|
|
home=tmp_path,
|
|
now=now,
|
|
trainer=in_process_trainer,
|
|
base_dir=tmp_path / "base",
|
|
)
|
|
assert res.promoted, res
|
|
state = load_train_state("main", tmp_path)
|
|
assert state.pre_promotion_downvote_rate == pytest.approx(0.2)
|
|
|
|
import json as _json
|
|
|
|
receipts = list((tmp_path / "router" / ".receipts").glob("main-*-promoted.json"))
|
|
assert receipts
|
|
payload = _json.loads(receipts[0].read_text())
|
|
assert payload["pre_promotion_downvote_rate"] == pytest.approx(0.2)
|
|
|
|
|
|
def test_orchestrator_no_feedback_baseline_is_none(tmp_path) -> None:
|
|
"""Zero recorded ratings -> unmeasured baseline (None), never 0.0."""
|
|
from datetime import UTC, datetime
|
|
|
|
from opensquilla.gateway.config import RouterSelfLearningConfig, SquillaRouterConfig
|
|
from opensquilla.squilla_router.self_learning.orchestrator import (
|
|
in_process_trainer,
|
|
maybe_run_update_router,
|
|
)
|
|
from opensquilla.squilla_router.self_learning.state import load_train_state
|
|
|
|
_write_min_store(tmp_path)
|
|
cfg = SquillaRouterConfig(
|
|
self_learning=RouterSelfLearningConfig(
|
|
enabled=True,
|
|
train_min_samples=4,
|
|
idle_hours=2.0,
|
|
cooldown_hours=72.0,
|
|
holdout_min_size=4,
|
|
holdout_pct=0.4,
|
|
holdout_repeats=2,
|
|
max_critical_under_routing=1.0,
|
|
cost_tolerance_pct=500.0,
|
|
)
|
|
)
|
|
(tmp_path / "base").mkdir() # empty base -> trains fresh
|
|
res = maybe_run_update_router(
|
|
"main",
|
|
router_cfg=cfg,
|
|
home=tmp_path,
|
|
now=datetime(2026, 6, 6, 12, 0, 0, tzinfo=UTC),
|
|
trainer=in_process_trainer,
|
|
base_dir=tmp_path / "base",
|
|
)
|
|
assert res.promoted, res
|
|
state = load_train_state("main", tmp_path)
|
|
assert state.pre_promotion_downvote_rate is None
|
|
# And with a None baseline the feedback trigger cannot fire even on a
|
|
# burst of early downvotes (should_rollback skips it).
|
|
from types import SimpleNamespace
|
|
|
|
from opensquilla.squilla_router.self_learning.promotion import should_rollback
|
|
|
|
assert not should_rollback(
|
|
pre_complaint_rate=0.5, # flat
|
|
post_complaint_rate=0.5,
|
|
post_n=100,
|
|
config=SimpleNamespace(),
|
|
pre_downvote_rate=None,
|
|
post_downvote_rate=1.0,
|
|
post_feedback_n=50,
|
|
)
|
|
|
|
|
|
def test_ensemble_downvotes_do_not_open_volume_gate(tmp_path) -> None:
|
|
"""Design invariant: gate counts only label-producing (single) downvotes."""
|
|
from opensquilla.squilla_router.self_learning.feedback import write_feedback
|
|
from opensquilla.squilla_router.self_learning.orchestrator import _with_feedback_stats
|
|
from opensquilla.squilla_router.self_learning.state import EventStoreStats
|
|
|
|
for i in range(6):
|
|
write_feedback(
|
|
"main",
|
|
decision_id=f"e-{i}",
|
|
session_key="agent:main:webchat:s1",
|
|
turn_index=i,
|
|
rating="down",
|
|
executed_kind="ensemble",
|
|
home=tmp_path,
|
|
)
|
|
stats = _with_feedback_stats(EventStoreStats(total=10), "main", tmp_path)
|
|
assert stats.feedback_down == 0 # ensemble downvotes produce no labels
|
|
|
|
write_feedback(
|
|
"main",
|
|
decision_id="s-1",
|
|
session_key="agent:main:webchat:s1",
|
|
turn_index=9,
|
|
rating="down",
|
|
executed_kind="single",
|
|
home=tmp_path,
|
|
)
|
|
stats2 = _with_feedback_stats(EventStoreStats(total=10), "main", tmp_path)
|
|
assert stats2.feedback_down == 1
|
|
|
|
|
|
def test_downvote_anchor_is_served_aware() -> None:
|
|
"""Gate/hold turns: the bump starts from the SERVED tier, not the raw
|
|
prediction, so a downvote never trains the tier the user just rejected."""
|
|
# Confidence gate: predicted R0, served R1 (gate default). Downvote must
|
|
# target R2, not R1.
|
|
gated = _sample(0, route="R0", final="R1", gate=True)
|
|
aligned = align_session([gated], _fbmap(0, "down"))
|
|
assert aligned[0].reason == REASON_EXPLICIT_DOWNVOTE
|
|
assert aligned[0].target_idx == 2
|
|
|
|
# Anti-downgrade hold: predicted R0, served R2. Served index is above the
|
|
# eligibility cap -> excluded, no upgrade label pointing below the served.
|
|
held = _sample(0, route="R0", final="R2")
|
|
aligned2 = align_session([held], _fbmap(0, "down"))
|
|
assert aligned2[0].reason == REASON_EXPLICIT_DOWNVOTE_HIGH_TIER
|
|
assert aligned2[0].target_idx == 2
|