""" Decision-detail endpoint, backed by public.event_decision (AI-extracted policy decisions). Serves a single decision by its unique key (event_decision_id), which the /search decision results link to (url=/decisions/{event_decision_id}). Mirrors the people.py person-detail pattern: a single fetchrow on the public serving table, 404 when absent. The pool registers no JSON codec, so asyncpg returns JSONB columns as raw JSON text — we json.loads() them here so vote_tally / competing_views / *_refs serialize to the client as real objects/arrays instead of double-encoded strings. """ from __future__ import annotations import json from datetime import date, timedelta from typing import Any, List, Optional from fastapi import APIRouter, HTTPException, Query from loguru import logger from opentelemetry import trace from pydantic import BaseModel from api.routes.search_postgres import get_db_pool router = APIRouter(prefix="/decision", tags=["decision"]) tracer = trace.get_tracer(__name__) class MeetingDocument(BaseModel): """An agenda/minutes PDF attached to the decision's meeting (suiteone).""" document_type: str url: str doc_date: Optional[str] = None # real DATE -> ISO 'YYYY-MM-DD' string on the wire body_name: Optional[str] = None class MeetingRecord(BaseModel): """The PARSED official agenda/minutes content for the decision's meeting (public.meeting_document, from the agenda/minutes enrichment). Distinct from MeetingDocument, which is just the document link/status.""" doc_kind: str # 'agenda' | 'minutes' meeting_body: Optional[str] = None agenda_items: Optional[Any] = None motions: Optional[Any] = None recorded_votes: Optional[Any] = None continuances: Optional[Any] = None legislation_numbers: Optional[Any] = None source_urls: Optional[Any] = None class DecisionDetail(BaseModel): """A single AI-extracted policy decision, keyed by event_decision_id.""" event_decision_id: str headline: Optional[str] = None decision_statement: Optional[str] = None outcome: Optional[str] = None primary_theme: Optional[str] = None # JSONB payloads (parsed from text); shapes vary by extraction. vote_tally: Optional[Any] = None human_element: Optional[Any] = None competing_views: Optional[Any] = None smart_brevity: Optional[Any] = None legislation_refs: Optional[Any] = None financial_item_refs: Optional[Any] = None place_refs: Optional[Any] = None # Location / provenance. jurisdiction_name: Optional[str] = None jurisdiction_type: Optional[str] = None state: Optional[str] = None state_code: Optional[str] = None city: Optional[str] = None c1_event_id: Optional[str] = None decision_id: Optional[str] = None subject_id: Optional[str] = None # Meeting context (event_meeting joined on c1_event_id). meeting_name: Optional[str] = None meeting_date: Optional[str] = None meeting_video_id: Optional[str] = None source_ai_model: Optional[str] = None extracted_at: Optional[str] = None # Associated meeting documents (agenda/minutes), joined on analysis_id -> # event_meeting_id. Empty list is a meaningful state (e.g. minutes unpublished). documents: List[MeetingDocument] = [] has_agenda: bool = False has_minutes: bool = False minutes_status: str = "not_published" # 'published' | 'not_published' # When minutes are unpublished: an ESTIMATED post date = meeting date + # the jurisdiction's median publish lag (jurisdiction_minutes_publish_lag). # Null when we have no reliable lag sample (sample_n < 5) — never fabricated. expected_minutes_date: Optional[str] = None # ISO 'YYYY-MM-DD' minutes_typical_lag_days: Optional[int] = None minutes_lag_sample_n: Optional[int] = None # Parsed official agenda/minutes content (public.meeting_document), joined to # the meeting by jurisdiction + date. Empty when none enriched for this meeting. meeting_record: List[MeetingRecord] = [] _DECISION_SQL = """ SELECT event_decision_id, headline, decision_statement, outcome, primary_theme, vote_tally, human_element, competing_views, smart_brevity, legislation_refs, financial_item_refs, place_refs, d.jurisdiction_name, d.jurisdiction_type, d.state, d.state_code, d.city, d.c1_event_id, d.analysis_id, d.decision_id, d.subject_id, d.source_ai_model, d.extracted_at, m.body_name AS meeting_name, CASE WHEN COALESCE(NULLIF(m.event_date, ''), NULLIF(m.meeting_date, '')) ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}' THEN COALESCE(NULLIF(m.event_date, ''), NULLIF(m.meeting_date, '')) END AS meeting_date, m.video_id AS meeting_video_id FROM event_decision d LEFT JOIN event_meeting m ON m.c1_event_id = d.c1_event_id WHERE d.event_decision_id = $1 """ # Parsed agenda/minutes content for a decision's meeting. Keyed by event_meeting_id # (= the decision's analysis_id) — the same clean join _DOCUMENTS_SQL uses, so it # aligns with the event_meeting_document link mart instead of a fragile # jurisdiction+date match. _MEETING_RECORD_SQL = """ SELECT md.doc_kind, md.meeting_body, md.agenda_items, md.motions, md.recorded_votes, md.continuances, md.legislation_numbers, md.source_urls FROM public.meeting_document md WHERE md.event_meeting_id = $1 ORDER BY md.doc_kind """ # JSONB columns asyncpg hands back as raw JSON text (no codec on the pool). _JSON_FIELDS = ( "vote_tally", "human_element", "competing_views", "smart_brevity", "legislation_refs", "financial_item_refs", "place_refs", ) def _parse_json(value: Any) -> Any: """asyncpg returns JSONB as text without a codec; tolerate already-parsed too.""" if value is None or not isinstance(value, str): return value try: return json.loads(value) except (json.JSONDecodeError, ValueError): return value # Meeting documents (agenda/minutes) for a decision's meeting. Joined on # event_meeting_id = the decision's analysis_id. Agenda first, then minutes, # then oldest doc_date first. Read-only fetch on a modeled public table. _DOCUMENTS_SQL = """ SELECT document_type, document_url, doc_date, body_name, census_geoid FROM event_meeting_document WHERE event_meeting_id = $1 ORDER BY CASE document_type WHEN 'agenda' THEN 0 WHEN 'minutes' THEN 1 ELSE 2 END, doc_date NULLS LAST """ # Typical minutes-publish lag for a jurisdiction, used to ESTIMATE when pending # minutes will post. Only jurisdictions with sample_n >= 5 have a row. _MINUTES_LAG_SQL = """ SELECT median_lag_days, sample_n FROM jurisdiction_minutes_publish_lag WHERE census_geoid = $1 """ # --------------------------------------------------------------------------- # Map drill-down: geocoded decision pins (nationwide, generic — no hardcoded # jurisdiction). Joins event_decision -> event_decision_place -> the geocoded # place view, returning one pin per (decision, plottable place). A decision with # multiple geocoded places yields multiple pins; we order is_primary places # first so the primary place leads. Caller can filter by state / bbox / theme. # --------------------------------------------------------------------------- _MAP_LIMIT_DEFAULT = 2_000 _MAP_LIMIT_MAX = 5_000 class DecisionMapPin(BaseModel): """A single geocoded civic-decision pin for the map.""" event_decision_id: str decision_id: Optional[str] = None place_id: str latitude: float longitude: float headline: Optional[str] = None primary_theme: Optional[str] = None outcome: Optional[str] = None # JSONB pass-through (parsed from text; asyncpg pool has no JSON codec). vote_tally: Optional[Any] = None jurisdiction_name: Optional[str] = None state_code: Optional[str] = None # ISO date string (from event_meeting on c1_event_id) or None. event_date: Optional[str] = None normalized_address: Optional[str] = None is_primary: bool = False class DecisionMapResponse(BaseModel): """Map payload: filters echoed back plus the plottable pins.""" state: Optional[str] = None theme: Optional[str] = None bbox: Optional[List[float]] = None total: int limit: int pins: List[DecisionMapPin] # Pin query. place_state_code is the place's own state (per spec). event_date is # pulled from event_meeting (text columns) joined on c1_event_id, matching the # detail route, and only surfaced when it looks like an ISO date. _MAP_SQL = """ SELECT d.event_decision_id, d.decision_id, g.place_id, g.latitude::float8 AS latitude, g.longitude::float8 AS longitude, d.headline, d.primary_theme, d.outcome, d.vote_tally, d.jurisdiction_name, g.place_state_code AS state_code, CASE WHEN COALESCE(NULLIF(m.event_date, ''), NULLIF(m.meeting_date, '')) ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}' THEN COALESCE(NULLIF(m.event_date, ''), NULLIF(m.meeting_date, '')) END AS event_date, g.normalized_address, COALESCE(edp.is_primary, FALSE) AS is_primary FROM event_decision d JOIN event_decision_place edp ON edp.event_decision_id = d.event_decision_id JOIN event_place_geocoded g ON g.place_id = edp.place_id LEFT JOIN event_meeting m ON m.c1_event_id = d.c1_event_id WHERE g.latitude IS NOT NULL AND g.longitude IS NOT NULL __FILTERS__ ORDER BY edp.is_primary DESC, d.event_decision_id, g.place_id LIMIT __LIMIT__ """ def _parse_bbox(bbox: Optional[str]) -> Optional[List[float]]: """Parse 'minLon,minLat,maxLon,maxLat' into 4 floats; 400 on bad input.""" if not bbox or not bbox.strip(): return None parts = [p.strip() for p in bbox.split(",")] if len(parts) != 4: raise HTTPException( status_code=400, detail="bbox must be 'minLon,minLat,maxLon,maxLat' (4 values)", ) try: min_lon, min_lat, max_lon, max_lat = (float(p) for p in parts) except ValueError as exc: raise HTTPException(status_code=400, detail="bbox values must be numbers") from exc if min_lon > max_lon or min_lat > max_lat: raise HTTPException( status_code=400, detail="bbox min must be <= max (order: minLon,minLat,maxLon,maxLat)", ) return [min_lon, min_lat, max_lon, max_lat] @router.get("/map", response_model=DecisionMapResponse) async def get_decision_map( state: Optional[str] = Query( None, min_length=2, max_length=2, description="2-letter state code; filters by the place's state", ), bbox: Optional[str] = Query( None, description="Viewport filter 'minLon,minLat,maxLon,maxLat'", ), theme: Optional[str] = Query( None, description="Filter by primary_theme (exact match)", ), limit: int = Query( _MAP_LIMIT_DEFAULT, ge=1, le=_MAP_LIMIT_MAX, description=f"Max pins (default {_MAP_LIMIT_DEFAULT}, cap {_MAP_LIMIT_MAX})", ), ) -> DecisionMapResponse: """ Geocoded civic-decision pins for a nationwide drill-down map. Joins event_decision -> event_decision_place -> event_place_geocoded and returns only plottable rows (latitude IS NOT NULL). One pin per (decision, geocoded place); primary places are ordered first. All filters are optional, supporting nation -> state -> viewport drill-down. Example:: GET /api/decision/map?state=AL&theme=Housing&limit=2000 """ state_code = state.strip().upper() if state else None bbox_vals = _parse_bbox(bbox) with tracer.start_as_current_span("decision-map") as span: span.set_attribute("decision_map.state", state_code or "") span.set_attribute("decision_map.theme", theme or "") span.set_attribute("decision_map.has_bbox", bbox_vals is not None) span.set_attribute("decision_map.limit", limit) filters: List[str] = [] params: List[Any] = [] idx = 1 if state_code: filters.append(f"AND g.place_state_code = ${idx}") params.append(state_code) idx += 1 if theme: filters.append(f"AND d.primary_theme = ${idx}") params.append(theme) idx += 1 if bbox_vals: min_lon, min_lat, max_lon, max_lat = bbox_vals filters.append( f"AND g.longitude BETWEEN ${idx} AND ${idx + 1} " f"AND g.latitude BETWEEN ${idx + 2} AND ${idx + 3}" ) params.extend([min_lon, max_lon, min_lat, max_lat]) idx += 4 limit_idx = idx params.append(limit) sql = _MAP_SQL.replace( "__FILTERS__", "\n ".join(filters) ).replace("__LIMIT__", f"${limit_idx}") try: pool = await get_db_pool() async with pool.acquire() as conn: with tracer.start_as_current_span("decision-map.query") as qspan: rows = await conn.fetch(sql, *params) qspan.set_attribute("decision_map.row_count", len(rows)) except Exception as e: # noqa: BLE001 — surface a clean 500 span.record_exception(e) logger.error("Decision map query failed: {}", e) raise HTTPException(status_code=500, detail="Failed to load decision map") pins = [ DecisionMapPin( event_decision_id=r["event_decision_id"], decision_id=r["decision_id"], place_id=r["place_id"], latitude=float(r["latitude"]), longitude=float(r["longitude"]), headline=r["headline"], primary_theme=r["primary_theme"], outcome=r["outcome"], vote_tally=_parse_json(r["vote_tally"]), jurisdiction_name=r["jurisdiction_name"], state_code=r["state_code"], event_date=r["event_date"], normalized_address=r["normalized_address"], is_primary=bool(r["is_primary"]), ) for r in rows ] span.set_attribute("decision_map.pin_count", len(pins)) logger.info( "🗺️ Decision map -> {} pins (state={}, theme={}, bbox={})", len(pins), state_code, theme, bbox_vals is not None, ) return DecisionMapResponse( state=state_code, theme=theme, bbox=bbox_vals, total=len(pins), limit=limit, pins=pins, ) @router.get("/{event_decision_id}", response_model=DecisionDetail) async def get_decision(event_decision_id: str) -> DecisionDetail: """ Return a single AI-extracted policy decision by event_decision_id. 404 if no decision row matches. """ with tracer.start_as_current_span("decision-detail") as span: span.set_attribute("decision.event_decision_id", event_decision_id) try: pool = await get_db_pool() doc_rows: list = [] lag_row = None # jurisdiction_minutes_publish_lag row (fetched within conn) async with pool.acquire() as conn: with tracer.start_as_current_span("decision-detail.query"): row = await conn.fetchrow(_DECISION_SQL, event_decision_id) if row is None: span.set_attribute("decision.found", False) raise HTTPException( status_code=404, detail=f"No decision found for event_decision_id '{event_decision_id}'", ) # Associated meeting documents (agenda/minutes). Defensive: a # missing/failing table or NULL analysis_id yields no documents, # never a 500 on the decision itself. analysis_id = row["analysis_id"] if analysis_id is not None: with tracer.start_as_current_span("decision-detail.documents") as doc_span: doc_span.set_attribute("decision.analysis_id", analysis_id) try: doc_rows = await conn.fetch(_DOCUMENTS_SQL, analysis_id) except Exception as doc_err: # noqa: BLE001 — documents are best-effort doc_span.record_exception(doc_err) logger.warning( "Meeting documents lookup failed for decision {} (analysis_id={}): {}", event_decision_id, analysis_id, doc_err, ) doc_rows = [] doc_span.set_attribute("decision.document_count", len(doc_rows)) # If minutes aren't among the docs, prefetch the # jurisdiction's publish-lag (keyed by the meeting's geoid) # so we can estimate when they'll post. Done here so it # runs while the connection is still held. has_minutes_pre = any(dr["document_type"] == "minutes" for dr in doc_rows) geoid_pre = next((dr["census_geoid"] for dr in doc_rows if dr["census_geoid"]), None) if not has_minutes_pre and geoid_pre: try: lag_row = await conn.fetchrow(_MINUTES_LAG_SQL, geoid_pre) except Exception as lag_err: # noqa: BLE001 — estimate is best-effort doc_span.record_exception(lag_err) logger.warning("Minutes-lag lookup failed (geoid={}): {}", geoid_pre, lag_err) # Parsed agenda/minutes content for this decision's meeting # (public.meeting_document). Best-effort — never 500 the decision. meeting_record_rows: list = [] if row["analysis_id"] is not None: try: meeting_record_rows = await conn.fetch(_MEETING_RECORD_SQL, row["analysis_id"]) except Exception as mr_err: # noqa: BLE001 — best-effort logger.warning( "meeting_record lookup failed for {}: {}", event_decision_id, mr_err ) span.set_attribute("decision.found", True) data = dict(row) data.pop("analysis_id", None) # internal join key, not part of the response shape for field in _JSON_FIELDS: data[field] = _parse_json(data.get(field)) extracted = data.get("extracted_at") data["extracted_at"] = extracted.isoformat() if extracted is not None else None documents: List[MeetingDocument] = [] for dr in doc_rows: dd = dr["doc_date"] documents.append( MeetingDocument( document_type=dr["document_type"], url=dr["document_url"], # Real DATE -> ISO string on the wire; tolerate text too. doc_date=dd.isoformat() if hasattr(dd, "isoformat") else (str(dd) if dd else None), body_name=dr["body_name"], ) ) data["documents"] = documents data["has_agenda"] = any(d.document_type == "agenda" for d in documents) has_minutes = any(d.document_type == "minutes" for d in documents) data["has_minutes"] = has_minutes data["minutes_status"] = "published" if has_minutes else "not_published" # Estimate when pending minutes will post: meeting date + the # jurisdiction's median publish lag (lag_row prefetched above). Only # when minutes are absent AND we know the meeting date AND the # jurisdiction has a reliable lag sample. Never fabricated. if not has_minutes and lag_row and lag_row["median_lag_days"] is not None: meeting_day = next( (dr["doc_date"] for dr in doc_rows if isinstance(dr["doc_date"], date)), None, ) if meeting_day is not None: median_lag = int(lag_row["median_lag_days"]) data["expected_minutes_date"] = ( meeting_day + timedelta(days=median_lag) ).isoformat() data["minutes_typical_lag_days"] = median_lag data["minutes_lag_sample_n"] = lag_row["sample_n"] data["meeting_record"] = [ MeetingRecord( doc_kind=r["doc_kind"], meeting_body=r["meeting_body"], agenda_items=_parse_json(r["agenda_items"]), motions=_parse_json(r["motions"]), recorded_votes=_parse_json(r["recorded_votes"]), continuances=_parse_json(r["continuances"]), legislation_numbers=_parse_json(r["legislation_numbers"]), source_urls=_parse_json(r["source_urls"]), ) for r in meeting_record_rows ] logger.info( "⚖️ Decision detail {} -> {} ({} docs, {} records)", event_decision_id, data.get("outcome"), len(documents), len(meeting_record_rows), ) return DecisionDetail(**data) except HTTPException: raise except Exception as e: span.record_exception(e) logger.error("Decision detail error for {}: {}", event_decision_id, e) raise HTTPException(status_code=500, detail="Failed to load decision detail") # --------------------------------------------------------------------------- # Related decisions — "similar" feed driven by OUR metadata (shared theme + same # body/jurisdiction), not by any video platform. Powers the decision page's # "Related decisions" rail. # --------------------------------------------------------------------------- class RelatedDecision(BaseModel): event_decision_id: str headline: Optional[str] = None jurisdiction_name: Optional[str] = None state_code: Optional[str] = None primary_theme: Optional[str] = None outcome: Optional[str] = None shared_theme: bool = False shared_jurisdiction: bool = False _RELATED_SQL = """ WITH base AS ( SELECT primary_theme, state_code, jurisdiction_name FROM event_decision WHERE event_decision_id = $1 ) SELECT e.event_decision_id, e.headline, e.jurisdiction_name, e.state_code, e.primary_theme, e.outcome, (e.primary_theme IS NOT DISTINCT FROM b.primary_theme) AS shared_theme, (e.jurisdiction_name IS NOT DISTINCT FROM b.jurisdiction_name) AS shared_jurisdiction, ( (e.primary_theme IS NOT DISTINCT FROM b.primary_theme)::int * 2 + (e.jurisdiction_name IS NOT DISTINCT FROM b.jurisdiction_name)::int * 2 + (e.state_code IS NOT DISTINCT FROM b.state_code)::int ) AS score FROM event_decision e, base b WHERE e.event_decision_id <> $1 AND ( (b.primary_theme IS NOT NULL AND e.primary_theme = b.primary_theme) OR (b.jurisdiction_name IS NOT NULL AND e.jurisdiction_name = b.jurisdiction_name) ) ORDER BY score DESC, e.extracted_at DESC NULLS LAST LIMIT $2 """ @router.get("/{event_decision_id}/related", response_model=List[RelatedDecision]) async def get_related_decisions( event_decision_id: str, limit: int = Query(6, ge=1, le=20), ) -> List[RelatedDecision]: """Decisions sharing this one's theme or body — ranked by overlap. [] if none.""" with tracer.start_as_current_span("decision-related") as span: span.set_attribute("decision.event_decision_id", event_decision_id) try: pool = await get_db_pool() async with pool.acquire() as conn: rows = await conn.fetch(_RELATED_SQL, event_decision_id, limit) except Exception as e: # noqa: BLE001 span.record_exception(e) logger.error("Related decisions error for {}: {}", event_decision_id, e) raise HTTPException(status_code=500, detail="Failed to load related decisions") return [ RelatedDecision( event_decision_id=r["event_decision_id"], headline=r["headline"], jurisdiction_name=r["jurisdiction_name"], state_code=r["state_code"], primary_theme=r["primary_theme"], outcome=r["outcome"], shared_theme=bool(r["shared_theme"]), shared_jurisdiction=bool(r["shared_jurisdiction"]), ) for r in rows ] # --------------------------------------------------------------------------- # Decision thread — the SAME item tracked ACROSS meetings (its lifecycle), so a # deferred item that returns later reads as one story instead of disconnected # rows. Links occurrences by a SPECIFIC place (a street/site place_id, never a # jurisdiction-wide stub that would falsely merge unrelated items) or by a # suffix-stripped subject slug (the trailing numeric id varies per meeting). # Ordered by meeting date; each row carries its outcome (so delays/continuances # show as steps) and the for/against view labels for an optional positions view. # --------------------------------------------------------------------------- class DecisionThreadItem(BaseModel): event_decision_id: str headline: Optional[str] = None outcome: Optional[str] = None primary_theme: Optional[str] = None meeting_name: Optional[str] = None meeting_date: Optional[str] = None meeting_video_id: Optional[str] = None is_current: bool = False # Optional for/against labels (from competing_views) for this occurrence. prevailing_label: Optional[str] = None counter_labels: List[str] = [] _THREAD_SQL = """ WITH base AS ( SELECT CASE WHEN starts_with(primary_place_id, 'place_') THEN primary_place_id END AS place_key, NULLIF(regexp_replace(COALESCE(subject_id, ''), '_[0-9]+$', ''), '') AS subj_key FROM event_decision WHERE event_decision_id = $1 ) SELECT d.event_decision_id, d.headline, d.outcome, d.primary_theme, d.competing_views, m.body_name AS meeting_name, CASE WHEN COALESCE(NULLIF(m.event_date, ''), NULLIF(m.meeting_date, '')) ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}' THEN COALESCE(NULLIF(m.event_date, ''), NULLIF(m.meeting_date, '')) END AS meeting_date, m.video_id AS meeting_video_id, (d.event_decision_id = $1) AS is_current FROM event_decision d LEFT JOIN event_meeting m ON m.c1_event_id = d.c1_event_id CROSS JOIN base b WHERE (b.place_key IS NOT NULL AND d.primary_place_id = b.place_key) OR (b.subj_key IS NOT NULL AND NULLIF(regexp_replace(COALESCE(d.subject_id, ''), '_[0-9]+$', ''), '') = b.subj_key) ORDER BY meeting_date NULLS LAST, d.extracted_at DESC NULLS LAST LIMIT 25 """ def _view_labels(competing_views: Any) -> tuple[Optional[str], List[str]]: """Pull (prevailing_label, [counter_labels]) out of a competing_views value. The pool has no JSONB codec, so the value arrives as TEXT — json.loads it. """ cv = competing_views if isinstance(cv, str): try: cv = json.loads(cv) except (ValueError, TypeError): return None, [] if not isinstance(cv, dict): return None, [] dom = cv.get("dominant_view") prevailing = dom.get("view_label") if isinstance(dom, dict) else None labels: List[str] = [] counters = cv.get("counter_views") if isinstance(counters, list): for c in counters: if isinstance(c, dict) and isinstance(c.get("view_label"), str): labels.append(c["view_label"]) return (prevailing if isinstance(prevailing, str) else None), labels @router.get("/{event_decision_id}/thread", response_model=List[DecisionThreadItem]) async def get_decision_thread(event_decision_id: str) -> List[DecisionThreadItem]: """ The same item across meetings (lifecycle), oldest → newest by meeting date. Returns a single self-row when the item appears in only one analyzed meeting; the client renders the cross-meeting timeline only when there are 2+. """ with tracer.start_as_current_span("decision-thread") as span: span.set_attribute("decision.event_decision_id", event_decision_id) try: pool = await get_db_pool() async with pool.acquire() as conn: rows = await conn.fetch(_THREAD_SQL, event_decision_id) except Exception as e: # noqa: BLE001 span.record_exception(e) logger.error("Decision thread error for {}: {}", event_decision_id, e) raise HTTPException(status_code=500, detail="Failed to load decision thread") items: List[DecisionThreadItem] = [] for r in rows: prevailing, counters = _view_labels(r["competing_views"]) items.append( DecisionThreadItem( event_decision_id=r["event_decision_id"], headline=r["headline"], outcome=r["outcome"], primary_theme=r["primary_theme"], meeting_name=r["meeting_name"], meeting_date=r["meeting_date"], meeting_video_id=r["meeting_video_id"], is_current=bool(r["is_current"]), prevailing_label=prevailing, counter_labels=counters, ) ) span.set_attribute("decision.thread_count", len(items)) return items