mirror of
https://github.com/fsecada01/Pygentic-AI.git
synced 2026-05-11 19:54:59 +00:00
feat: implement HTMX OOB swaps for incremental status updates
Replace full container refresh with incremental status item appends using HTMX Out-of-Band swaps for smoother UX without flashing. Problem: - Status container innerHTML swap replaced entire timeline every second - All items re-rendered causing visual flashing even with CSS transitions - Inefficient: re-sends all messages every poll Solution: - Track last message index per session - Return only NEW messages since last poll - Use HTMX hx-swap-oob="beforeend:#status-timeline" to append items - First poll returns full container, subsequent polls return only deltas Architecture: 1. Session Tracking (consts.py): - Added last_message_index dict to track per-session state - Memory-based, resets on server restart (acceptable for dev) 2. New Template (status_item.html): - Single status article with OOB swap attribute - Reusable component for incremental updates 3. Backend Logic (router.py): - First poll: full status.html with all messages - Subsequent: only new messages with OOB attributes - Empty response if no new messages 4. Frontend (home.html): - Removed settle:200ms (no longer needed) - Status timeline receives appended items Benefits: - Eliminates flashing during updates - More efficient (only sends deltas) - Smoother perceived performance - No JavaScript state management required - Automatic HTMX OOB handling Technical Details: - OOB swap: Items append to #status-timeline automatically - Session index tracks progress per analysis - First poll initializes container structure - Graceful degradation if session tracking resets Testing: - Verify first poll shows container + all messages - Verify subsequent polls append only new items - Verify no flashing during status updates - Verify empty responses when no new messages Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
@ -7,3 +7,5 @@ ANALYSIS_COMPLETE_MESSAGE = "Analysis complete!"
|
||||
running_tasks = set()
|
||||
status_store: dict[str, list] = defaultdict(list)
|
||||
result_store: dict[str, Any] = {}
|
||||
# Track last message count seen by each session for incremental updates
|
||||
last_message_index: dict[str, int] = defaultdict(int)
|
||||
|
||||
@ -12,6 +12,7 @@ from backend.settings import app_settings
|
||||
from backend.site.consts import (
|
||||
ANALYSIS_COMPLETE_MESSAGE,
|
||||
ANALYZING_MESSAGE,
|
||||
last_message_index,
|
||||
result_store,
|
||||
running_tasks,
|
||||
status_store,
|
||||
@ -91,26 +92,51 @@ async def analyze_url(
|
||||
@user_frontend.get("/status", response_class=HTMLResponse)
|
||||
async def get_status(request: Request):
|
||||
"""
|
||||
Returns the current status messages
|
||||
Returns new status messages since last poll (incremental updates)
|
||||
First poll returns all messages + container, subsequent polls return
|
||||
only new items with HTMX OOB swap
|
||||
:param request:
|
||||
:return:
|
||||
"""
|
||||
context = {"request": request, "messages": [], "result": False}
|
||||
session_id = request.session.get("analysis_id")
|
||||
if session_id:
|
||||
# logger.info(f"Found session id! {session_id}")
|
||||
messages = status_store.get(session_id, [])
|
||||
result = ANALYSIS_COMPLETE_MESSAGE in messages
|
||||
# logger.info(
|
||||
# f"Status check - Session ID: {session_id}, Messages: "
|
||||
# f"{messages}",
|
||||
# )
|
||||
if not session_id:
|
||||
return templates.TemplateResponse(
|
||||
"status.html", {"request": request, "messages": [], "result": False}
|
||||
)
|
||||
|
||||
context.update({"messages": messages, "result": result})
|
||||
all_messages = status_store.get(session_id, [])
|
||||
last_index = last_message_index.get(session_id, 0)
|
||||
result = ANALYSIS_COMPLETE_MESSAGE in all_messages
|
||||
|
||||
# logger.info(context)
|
||||
# First poll: return full container with all messages
|
||||
if last_index == 0:
|
||||
last_message_index[session_id] = len(all_messages)
|
||||
return templates.TemplateResponse(
|
||||
"status.html",
|
||||
{"request": request, "messages": all_messages, "result": result},
|
||||
)
|
||||
|
||||
return templates.TemplateResponse("status.html", context=context)
|
||||
# Subsequent polls: return only new messages with OOB swap
|
||||
new_messages = all_messages[last_index:]
|
||||
last_message_index[session_id] = len(all_messages)
|
||||
|
||||
if not new_messages:
|
||||
# No new messages, return empty response
|
||||
return HTMLResponse(content="", status_code=200)
|
||||
|
||||
# Render new items with OOB swap
|
||||
items_html = ""
|
||||
for idx, message in enumerate(new_messages):
|
||||
is_last = idx == len(new_messages) - 1
|
||||
item = templates.get_template("status_item.html").render(
|
||||
message=message,
|
||||
is_last=is_last,
|
||||
result=result,
|
||||
request=request,
|
||||
)
|
||||
items_html += item
|
||||
|
||||
return HTMLResponse(content=items_html, status_code=200)
|
||||
|
||||
|
||||
@user_frontend.get("/result", response_class=HTMLResponse)
|
||||
|
||||
@ -102,13 +102,13 @@
|
||||
|
||||
<section class="section"
|
||||
id="analysis-content">
|
||||
<!-- Status Updates -->
|
||||
<!-- Status Updates (OOB Swap Target) -->
|
||||
<div x-show="analyzing && !hasResults"
|
||||
x-transition
|
||||
id="status"
|
||||
hx-get={{ url_for('get_status') }}
|
||||
hx-trigger="every 1s[!document.querySelector('#result-container')]"
|
||||
hx-swap="innerHTML settle:200ms"
|
||||
hx-swap="innerHTML"
|
||||
@htmx:after-request="
|
||||
const response = $event.detail.xhr.response.trim();
|
||||
if (response && response.length > 0) {
|
||||
|
||||
32
src/frontend/templates/status_item.html
Normal file
32
src/frontend/templates/status_item.html
Normal file
@ -0,0 +1,32 @@
|
||||
{# Single status item template for HTMX OOB swaps #}
|
||||
{% set is_error = message.startswith('Error:') %}
|
||||
{% set is_loading = is_last and not result %}
|
||||
{% set is_tool_message = message.startswith('Using tool') %}
|
||||
{% set is_complete = "complete" in message.lower() or "done" in message.lower() %}
|
||||
{% set status_label = 'Error' if is_error else ('Complete' if is_complete else ('In Progress' if is_loading else ('Processing' if is_tool_message else 'Status'))) %}
|
||||
|
||||
<article class="status-item {% if is_error %}status-item--error{% elif is_complete %}status-item--success{% elif is_loading %}status-item--loading{% else %}status-item--info{% endif %}"
|
||||
role="article"
|
||||
aria-label="{{ status_label }}: {{ message }}"
|
||||
hx-swap-oob="beforeend:#status-timeline">
|
||||
<div class="status-item__indicator"
|
||||
aria-hidden="true">
|
||||
{% if is_error %}
|
||||
<i class="fas fa-circle-xmark"></i>
|
||||
{% elif is_complete %}
|
||||
<i class="fas fa-circle-check"></i>
|
||||
{% elif is_loading %}
|
||||
<i class="fas fa-circle-notch fa-spin"></i>
|
||||
{% else %}
|
||||
<i class="fas fa-circle"></i>
|
||||
{% endif %}
|
||||
</div>
|
||||
<div class="status-item__content">
|
||||
<div class="status-item__header">
|
||||
{{ status_label }}
|
||||
</div>
|
||||
<div class="status-item__message">
|
||||
{{ message }}
|
||||
</div>
|
||||
</div>
|
||||
</article>
|
||||
Reference in New Issue
Block a user