[Artifacts] Dashboard data source refresh engine done coding:7 reasoning:6

[Artifacts] Dashboard data source refresh engine

Goal

Dashboards need an engine that periodically evaluates their data sources and re-renders the template HTML with fresh values. This is the "live" part of living dashboards — without it, dashboards are just static HTML.

Data Source Evaluation

sql_query

def evaluate_sql_query(source: dict) -> str:
    """Execute SQL against PostgreSQL and format results."""
    db = sqlite3.connect('postgresql://scidex', timeout=10.0)
    db.row_factory = sqlite3.Row
    rows = db.execute(source['query']).fetchall()
    db.close()
    
    if len(rows) == 1 and len(rows[0].keys()) == 1:
        # Single value — return as string
        return str(rows[0][0])
    else:
        # Multiple rows — return as HTML table or JSON
        return format_as_html_table(rows)

Safety: Only SELECT queries allowed. Queries are sanitized to prevent writes. Timeout of 10 seconds per query.

artifact_ref

def evaluate_artifact_ref(source: dict) -> str:
    """Fetch latest version of an artifact and format for display."""
    artifact = get_artifact(source['artifact_id'])
    if not artifact:
        return '<span class="missing">Artifact not found</span>'
    view = source.get('view', 'summary')
    return render_artifact_embed(source['artifact_id'], view)

api_endpoint

def evaluate_api_endpoint(source: dict) -> str:
    """Fetch from internal API endpoint."""
    import urllib.request, json
    url = f"http://localhost:8000{source['url']}"
    response = urllib.request.urlopen(url, timeout=5)
    data = json.loads(response.read())
    if source.get('jq_filter'):
        data = apply_jq_filter(data, source['jq_filter'])
    return format_value(data)

Safety: Only localhost:8000 endpoints allowed. No external HTTP requests.

kg_query

def evaluate_kg_query(source: dict) -> str:
    """Query knowledge graph and format results."""
    # Use existing KG query infrastructure
    results = query_knowledge_graph(source['query'])
    return format_kg_results(results)

Refresh Pipeline

def refresh_dashboard(dashboard_artifact_id: str) -> str:
    """Re-render a dashboard with fresh data."""
    artifact = get_artifact(dashboard_artifact_id)
    metadata = json.loads(artifact['artifact']['metadata'])
    template = metadata['template_html']
    
    # Evaluate each data source
    rendered = template
    for source in metadata['data_sources']:
        evaluator = DATA_SOURCE_EVALUATORS.get(source['type'])
        if evaluator:
            value = evaluator(source)
            rendered = rendered.replace(f"{{{{data:{source['key']}}}}}", value)
    
    # Also resolve any artifact embeds
    rendered = resolve_embeds(rendered)
    
    # Cache rendered HTML
    metadata['rendered_html'] = rendered
    metadata['last_rendered_at'] = datetime.utcnow().isoformat()
    update_artifact_metadata(dashboard_artifact_id, metadata)
    
    return rendered

Acceptance Criteria

Dependencies

• a17-28-DASH0001 (dashboard artifact type and metadata schema)
• a17-29-EMBD0001 (artifact embed rendering for embedded artifacts in dashboards)

Dependents

• d16-26-DDSH0001 (demo: knowledge growth dashboard)

Work Log

2026-04-26 04:40 UTC — Slot a17-30-DREF0001

• Read AGENTS.md, dashboard_engine.py, spec, api.py routes
• Explored existing infrastructure: render_dashboard() for view_spec_json, SQL-only; knowledge_edges table schema confirmed
• Implemented in scidex/senate/dashboard_engine.py:

- _format_as_html_table(rows) — HTML table from row dicts
- _format_value(data) — JSON/string formatter
- evaluate_sql_query(source) — uses existing execute_safe_query() with SELECT-only safety + 10s timeout
- evaluate_artifact_ref(source) — fetches artifact via get_db(), returns embed HTML card (nested dashboard renders its own rendered_html)
- evaluate_api_endpoint(source) — GET localhost:8000 only, 5s timeout, optional dot-path jq_filter
- evaluate_kg_query(source) — searches knowledge_edges by ILIKE on source_id/target_id, 10s timeout
- DATA_SOURCE_EVALUATORS dispatch dict
- DATA_PLACEHOLDER_RE — regex for {{data:KEY}} placeholders
- refresh_dashboard(dashboard_artifact_id) — main pipeline: loads template_html, evaluates data_sources, substitutes {{data:KEY}}, calls resolve_embeds(), persists rendered_html + last_rendered_at
- Added knowledge_edges to ALLOWED_TABLES

• Updated api.py /api/dashboard/{id}/refresh endpoint to route through refresh_dashboard() when template_html is present, fall back to render_dashboard() for view_spec_json dashboards
• Tested: SQL query returns 1351 (correct count), KG query returns edges, artifact_ref returns embed HTML
• End-to-end: refresh_dashboard() with template_html + sql_query correctly substituted {{data:count}} → 1351

Acceptance Criteria Status