radionAI / session.py
htgudi's picture
deploy 6bcac61
5bb3295
Raw History Blame Contribute Delete
4.06 kB
"""
Per-session run history — YashTech Ltd. AI Labs.
Holds every response produced in one browser session so that a second run does
not silently discard the first, and so an export can carry the whole session
rather than whichever pane happens to be on screen.
**Where this lives, and why it matters.** The log is Gradio session state:
server-side, in memory, scoped to one browser tab, and gone when that tab is
closed or reloaded. It is deliberately NOT browser storage.
A generated report can carry a patient pseudonym, a clinical indication and
findings. `localStorage` would put that on the workstation's disk, outside the
30-minute temp pruning, surviving logout, readable by any later user of that
machine, and outside anything the DPIA covers. Losing a session's history on
refresh is an inconvenience; leaving clinical text on a shared workstation is
an incident. See COMPLIANCE.md A1/A3.
"""
from __future__ import annotations
import datetime as _dt
MAX_ENTRIES = 50 # a session that runs longer than this is a new session
def append(log, *, task_id: str, task_label: str, text: str,
question: str = "", system_prompt: str = "", now=None):
"""Return a NEW log with this run appended.
Gradio compares state by identity for change detection, so the list is
replaced rather than mutated — an in-place append would not propagate.
Empty or refusal-only output is not recorded: a session history full of
'Sign in to run tasks' is noise.
"""
log = list(log or [])
body = (text or "").strip()
if not body:
return log
stamp = (now or _dt.datetime.now(_dt.timezone.utc)).strftime("%H:%M:%S")
log.append({
"n": len(log) + 1,
"time": stamp,
"task_id": task_id,
"task_label": task_label or task_id,
"question": (question or "").strip(),
"system_prompt": (system_prompt or "").strip(),
"text": body,
})
return log[-MAX_ENTRIES:]
def summary(log) -> str:
"""One line for the accordion label, so a collapsed panel still says what is in it."""
entries = list(log or [])
if not entries:
return "Session history — nothing run yet"
tasks = sorted({e["task_label"] for e in entries})
shown = ", ".join(tasks[:3]) + (f" +{len(tasks) - 3} more" if len(tasks) > 3 else "")
return f"Session history — {len(entries)} run(s): {shown}"
def render(log) -> str:
"""Markdown for the history panel, newest first."""
entries = list(log or [])
if not entries:
return (
"_Nothing run in this session yet._\n\n"
"Each run is kept here so a second query does not discard the first. "
"This history lives in the session only — it is cleared when the page "
"is reloaded or the tab is closed, and it is never written to disk."
)
out = []
for e in reversed(entries):
out.append(f"### {e['n']}. {e['task_label']} · {e['time']} UTC")
if e["question"]:
out.append(f"> {e['question']}")
out.append("")
out.append(e["text"])
out.append("\n---\n")
return "\n".join(out)
def combined(log) -> str:
"""Every run in the session, oldest first, for a whole-session export."""
entries = list(log or [])
if not entries:
return ""
out = [f"_{len(entries)} run(s) in this session._", ""]
for e in entries:
out.append(f"## {e['n']}. {e['task_label']}")
out.append("")
out.append(f"- **Task:** `{e['task_id']}`")
out.append(f"- **Run at:** {e['time']} UTC")
if e["question"]:
out.append(f"- **Question:** {e['question']}")
out.append("")
out.append(e["text"])
if e["system_prompt"]:
out += ["", "<details><summary>System prompt used</summary>", "",
f"> {e['system_prompt']}", "", "</details>"]
out.append("\n---\n")
return "\n".join(out)
def tasks_in(log) -> list[str]:
return sorted({e["task_id"] for e in (log or [])})