Function bodies 688 total

Name: Aljefra Mapper analysis
Creator: Repobility
License: https://repobility.com/legal/terms/

_run_evals function · python · L712-L834 (123 LOC)

src/pinwheel/core/game_loop.py

async def _run_evals(
    repo: Repository,
    season_id: str,
    round_number: int,
    reports: list[Report],
    game_summaries: list[dict],
    teams_cache: dict,
    api_key: str = "",
) -> None:
    """Run automated evals after report generation. Non-blocking."""
    from pinwheel.evals.behavioral import compute_report_impact_rate
    from pinwheel.evals.grounding import GroundingContext, check_grounding
    from pinwheel.evals.prescriptive import scan_prescriptive

    # Build grounding context
    team_data = [{"name": t.name} for t in teams_cache.values()]
    hooper_data = []
    for t in teams_cache.values():
        for h in t.hoopers:
            hooper_data.append({"name": h.name})
    season = await repo.get_season(season_id)
    ruleset_dict = (season.current_ruleset if season else None) or {}
    context = GroundingContext(
        team_names=[d["name"] for d in team_data],
        agent_names=[d["name"] for d in hooper_data],
        rule_params=list((ruleset_dict o

tally_pending_governance function · python · L837-L1037 (201 LOC)

src/pinwheel/core/game_loop.py

async def tally_pending_governance(
    repo: Repository,
    season_id: str,
    round_number: int,
    ruleset: RuleSet,
    event_bus: EventBus | None = None,
    effect_registry: EffectRegistry | None = None,
    meta_store: MetaStore | None = None,
    skip_deferral: bool = False,
) -> tuple[RuleSet, list[VoteTally], dict]:
    """Tally all pending proposals and enact passing rule changes.

    Standalone function — can run with or without game simulation.
    When ``effect_registry`` is provided, passing proposals that contain
    v2 effects (meta_mutation, hook_callback, narrative) will have those
    effects registered in the registry and persisted to the event store.
    When ``meta_store`` is provided, fires ``gov.pre`` and ``gov.post``
    hooks around the governance tally for any registered effects.
    When ``skip_deferral`` is True, the minimum voting period is bypassed
    (used for season-close catch-up tallies).
    Returns (updated_ruleset, tallies, governance_data).

_get_series_games function · python · L2084-L2114 (31 LOC)

src/pinwheel/core/game_loop.py

async def _get_series_games(
    repo: Repository,
    season_id: str,
    team_a_id: str,
    team_b_id: str,
) -> list[dict]:
    """Get all playoff games between two teams, ordered chronologically.

    Returns a list of dicts with home/away team ids, scores, round number.
    """
    playoff_schedule = await repo.get_full_schedule(season_id, phase="playoff")
    playoff_rounds = {s.round_number for s in playoff_schedule}
    all_games = await repo.get_all_games(season_id)

    pair = frozenset({team_a_id, team_b_id})
    series_games: list[dict] = []
    for g in sorted(all_games, key=lambda g: (g.round_number, g.matchup_index)):
        if g.round_number not in playoff_rounds:
            continue
        if frozenset({g.home_team_id, g.away_team_id}) == pair:
            series_games.append(
                {
                    "round_number": g.round_number,
                    "home_team_id": g.home_team_id,
                    "away_team_id": g.away_team_id,

_generate_series_reports function · python · L2117-L2313 (197 LOC)

src/pinwheel/core/game_loop.py

async def _generate_series_reports(
    repo: Repository,
    season_id: str,
    deferred_events: list[tuple[str, dict]],
    teams_cache: dict[str, Team],
    api_key: str = "",
) -> list[dict]:
    """Generate series reports for any completed series in deferred events.

    Scans deferred_events for ``season.semifinals_complete`` and
    ``season.playoffs_complete`` events, gathers game data for each
    completed series, generates an AI recap, and stores it.

    Returns list of report event dicts for downstream publishing.
    """
    report_events: list[dict] = []

    for event_type, event_data in deferred_events:
        if event_type == "season.semifinals_complete":
            # Each semi series that just completed
            for semi in event_data.get("semi_series", []):
                winner_id = semi.get("winner_id", "")
                loser_id = semi.get("loser_id", "")
                if not winner_id or not loser_id:
                    continue

                winn

step_round function · python · L2797-L2872 (76 LOC)

src/pinwheel/core/game_loop.py

async def step_round(
    repo: Repository,
    season_id: str,
    round_number: int,
    event_bus: EventBus | None = None,
    api_key: str = "",
    governance_interval: int = 1,
    suppress_spoiler_events: bool = False,
) -> RoundResult:
    """Execute one complete round of the game loop.

    Returns a RoundResult with game results, governance outcomes, and reports.
    Delegates to the three phase functions but keeps everything in one session
    (backward-compatible single-session behavior).
    """
    start = time.monotonic()
    logger.info("round_start season=%s round=%d", season_id, round_number)

    phase1_start = time.perf_counter()
    sim = await _phase_simulate_and_govern(
        repo,
        season_id,
        round_number,
        event_bus=event_bus,
        governance_interval=governance_interval,
        suppress_spoiler_events=suppress_spoiler_events,
    )
    phase1_ms = (time.perf_counter() - phase1_start) * 1000
    logger.info(
        "phase_timing pha

step_round_multisession function · python · L2875-L2973 (99 LOC)

src/pinwheel/core/game_loop.py

async def step_round_multisession(
    engine: object,
    season_id: str,
    round_number: int,
    event_bus: EventBus | None = None,
    api_key: str = "",
    governance_interval: int = 1,
    suppress_spoiler_events: bool = False,
) -> RoundResult:
    """Execute one round with separate DB sessions per phase.

    Releases the SQLite write lock between phases so Discord commands
    (/join, /propose, /vote) can write freely during slow AI calls.

    Lock timeline:
        Session 1 (~2-3s): simulate games, store results, tally governance
           [LOCK RELEASED]
        AI calls (~30-90s): commentary, highlights, reports (NO session open)
           [LOCK RELEASED]
        Session 2 (~1-2s): store reports, run evals, season progression
           [LOCK RELEASED]

    The ``engine`` parameter is typed as ``object`` to avoid importing
    AsyncEngine at module level; callers pass an ``AsyncEngine`` instance.
    """
    from pinwheel.db.engine import get_session as _get_session

RoundResult.__init__ method · python · L2979-L3011 (33 LOC)

src/pinwheel/core/game_loop.py

    def __init__(
        self,
        round_number: int,
        games: list[dict],
        reports: list[Report],
        tallies: list[VoteTally],
        game_results: list[GameResult] | None = None,
        game_row_ids: list[str] | None = None,
        teams_cache: dict | None = None,
        governance_summary: dict | None = None,
        season_complete: bool = False,
        final_standings: list[dict] | None = None,
        playoff_bracket: list[dict] | None = None,
        playoffs_complete: bool = False,
        finals_matchup: dict | None = None,
        report_events: list[dict] | None = None,
        deferred_season_events: list[tuple[str, dict]] | None = None,
    ) -> None:
        self.round_number = round_number
        self.games = games
        self.reports = reports
        self.tallies = tallies
        self.game_results = game_results or []
        self.game_row_ids = game_row_ids or []
        self.teams_cache = teams_cache or {}
        self.governance_summar

Repobility's GitHub App fixes findings like these · https://github.com/apps/repobility-bot

remove_invisible_chars function · python · L48-L66 (19 LOC)