import json
import os
import re
import sys
import time
import html
import urllib.error
import urllib.request
import xml.etree.ElementTree as ET
from pathlib import Path
from typing import Any, Optional
import streamlit as st
try:
from huggingface_hub import HfApi
except Exception:
HfApi = None
SRC = Path(__file__).resolve().parent
REPO_ROOT = SRC.parent
for extra in (SRC, REPO_ROOT / "src"):
extra_str = str(extra)
if extra_str not in sys.path:
sys.path.insert(0, extra_str)
import runner as runner_module
from runner import PipelineConfig
from common.paper_package import load_paper_package
from step_08_annotation.pipeline import TwoPassAnnotationPipeline
from streamlit_config import EXAMPLES, TAB_NAMES
import replay as replay_module
import system_run_data as system_run_module
import system_live_runner as system_live_module
from workflow_graph import STEP_COPY, WORK_PANEL_STEPS
try:
from streamlit_config import APP_NAV, APP_VIEWS
except ImportError:
APP_VIEWS = [
"Annotation Process",
"System Run",
]
APP_NAV = [
{"view": "Annotation Process", "icon": ":material/account_tree:"},
{"view": "System Run", "icon": ":material/play_circle:"},
]
DEFAULT_SOURCE_ROOT = str(REPO_ROOT / "src" / "processed_papers")
DEFAULT_OUTPUT_ROOT = str(REPO_ROOT / "hf_space" / "runs")
REPLAY_STEP_DELAY_SEC = float(os.getenv("REPLAY_STEP_DELAY_SEC", "0.35"))
# Extra hold after a step completes so the workflow pulse is readable.
REPLAY_STEP_HOLD_SEC = float(os.getenv("REPLAY_STEP_HOLD_SEC", "0.55"))
DEFAULT_APP_VIEW = APP_VIEWS[0]
CUSTOM_CSS = """
"""
def get_secret(name: str, default: str = "") -> str:
value = os.getenv(name)
if value:
return value
try:
return st.secrets[name]
except Exception:
return default
def run_repo_config() -> tuple[str | None, str, str | None]:
repo_id = get_secret("RUNS_REPO_ID", "")
repo_type = get_secret("RUNS_REPO_TYPE", "dataset")
token = get_secret("HF_WRITE_TOKEN", "") or get_secret("HF_TOKEN", "")
return repo_id or None, repo_type, token or None
def remote_run_prefix(job_id: str) -> str:
return f"runs/{job_id}"
def upload_run_artifact(job_dir: Path) -> str:
repo_id, repo_type, token = run_repo_config()
if not repo_id or not token:
return ""
if HfApi is None:
return "upload_failed: huggingface_hub is not installed"
job_id = job_dir.name
remote_prefix = remote_run_prefix(job_id)
uploaded: list[str] = []
try:
api = HfApi(token=token)
for name in ["input_ids.json", "run_config.json", "summary.txt"]:
path = job_dir / name
if path.exists():
api.upload_file(
path_or_fileobj=str(path),
path_in_repo=f"{remote_prefix}/{name}",
repo_id=repo_id,
repo_type=repo_type,
commit_message=f"Upload {name} for {job_id}",
)
uploaded.append(name)
for folder_name in ["logs", "processed_papers", "two_pass_outputs"]:
folder = job_dir / folder_name
if not folder.exists():
continue
files = [path for path in folder.rglob("*") if path.is_file()]
if not files:
continue
api.upload_folder(
folder_path=str(folder),
path_in_repo=f"{remote_prefix}/{folder_name}",
repo_id=repo_id,
repo_type=repo_type,
commit_message=f"Upload {folder_name} for {job_id}",
ignore_patterns=["__pycache__/*", "*.pyc", "*.zip"],
)
uploaded.append(f"{folder_name}[{len(files)} files]")
return f"{repo_type}:{repo_id}/{remote_prefix}/ (uploaded: {', '.join(uploaded) or 'nothing'})"
except Exception as exc:
return f"upload_failed: {exc}"
def _load_json(path: Path) -> Optional[dict]:
if not path.exists():
return None
try:
return json.loads(path.read_text(encoding="utf-8"))
except Exception:
return None
def _status_from_line(line: str, current: str) -> str:
text = (line or "").strip()
text = _display_log_line(text)
if text.startswith("Pipeline stopped:"):
return "Stopped"
if text.startswith("Step "):
return text
if "failed" in text.lower():
return f"Failed: {text}"
if "completed successfully" in text.lower():
return "Completed"
return current
def _display_log_line(line: str) -> str:
text = (line or "").strip()
if text.startswith("Step ") and " failed." in text:
return text.splitlines()[0]
if text == "[annotation] starting cluster-first two-pass annotation":
return f"Step 8/8: {STEP_COPY[8]}"
if text.startswith("[annotation] complete:"):
return "Step 8 complete"
# Rewrite engineering step labels to the shared audience-facing copy.
m = re.match(r"Step\s+(\d+)\s*/\s*\d+\s*:\s*(.*)$", text)
if m:
n = int(m.group(1))
if n in STEP_COPY:
return f"Step {n}/8: {STEP_COPY[n]}"
if text == "Pipeline completed successfully.":
return text
return text
def _resolve_graph_sources(paper_input: str) -> tuple[Optional[Path], Optional[dict]]:
"""Paper dir + annotation payload for the workflow graph (trace or session)."""
entry = _paper_result_for_input(paper_input)
if entry:
paper_dir_path = entry.get("paper_dir_path")
payload_path = entry.get("annotation_payload_path")
paper_dir = Path(paper_dir_path) if paper_dir_path else None
payload = _load_json(Path(payload_path)) if payload_path else None
if paper_dir and paper_dir.exists():
return paper_dir, payload if isinstance(payload, dict) else None
trace = replay_module.load_trace(paper_input)
if not trace:
return None, None
paper_dir_path = trace.get("paper_dir_path")
payload_path = trace.get("annotation_payload_path")
paper_dir = Path(paper_dir_path) if paper_dir_path else None
payload = _load_json(Path(payload_path)) if payload_path else None
return paper_dir, payload if isinstance(payload, dict) else None
def _workflow_run_id(paper_input: str) -> str:
arxiv_id = replay_module.parse_arxiv_id(paper_input or "") or "default"
return f"scipaths:{arxiv_id}"
def _mount_graph_shell(graph_slot) -> None:
"""Mount the graph canvas shell (st.iframe HTML srcdoc + tree layout JS)."""
if graph_slot is None:
return
# Call-time import; avoid importlib.reload so identical shell HTML can be
# reused across Streamlit runs instead of forcing a fresh iframe each time.
from workflow_graph import render_graph_shell
render_graph_shell(graph_slot)
def _push_graph(courier_slot, paper_input: str, events: list[str]) -> None:
"""Push an incremental graph model into the mounted shell via postMessage.
This never re-emits the graph iframe, so the canvas grows in place instead of
reloading on every step.
"""
if courier_slot is None:
return
paper_dir, payload = _resolve_graph_sources(paper_input)
render_events = list(events or [])
if not render_events and paper_dir is not None:
# Idle selected paper: show target only (no active pulse).
render_events = ["Step 1/8: Load the paper", "Step 1 complete"]
# Import at call-time so Streamlit always picks up workflow_graph changes.
from workflow_graph import build_synced_model, push_graph_update
model = build_synced_model(
paper_dir=paper_dir,
payload=payload,
events=render_events,
run_id=_workflow_run_id(paper_input),
)
push_graph_update(courier_slot, model)
def _format_step_event(line: str) -> str:
text = _display_log_line(line)
if not text:
return ""
return text
STATE_DEFAULTS = {
"paper_input": "",
"run_status": "Idle",
"run_logs": [],
"run_events": [],
"artifact_path": None,
"run_dir_path": None,
"paper_dir_path": None,
"annotation_payload_path": None,
"run_summary": None,
"annotation_skipped_reason": None,
"pipeline_stopped_reason": None,
"pipeline_failed_reason": None,
"remote_artifact_ref": "",
"replay_mode": False,
"live_run_mode": False,
# Locked demo selection (survives run clears / post-run reruns).
"selected_demo_url": "",
# System Run case-study selection (independent of Annotation Process).
"system_run_mode": "prerun", # "prerun" | "live"
"selected_system_method": system_run_module.DEFAULT_SYSTEM_METHOD,
"selected_system_case": system_run_module.DEFAULT_SYSTEM_CASE,
"selected_live_method": "codeagent_parametric",
# Per-method Live Run state (parametric vs websearch_deep are independent).
"live_run_by_method": {},
"app_view": DEFAULT_APP_VIEW,
# Per-paper result cache so Clusters/Decomposition follow the selected paper.
"run_results_by_paper": {},
}
# Widget keys owned by Streamlit; assign defaults instead of deleting them.
WIDGET_STATE_KEYS = {"paper_input"}
# Survives per-run clears; only wiped by Reset session.
PRESERVED_STATE_KEYS = {
"app_view",
"run_results_by_paper",
"live_run_mode",
"selected_demo_url",
"system_run_mode",
"selected_system_method",
"selected_system_case",
"selected_live_method",
"live_run_by_method",
}
def _copy_default(value: Any) -> Any:
if isinstance(value, list):
return list(value)
if isinstance(value, dict):
return dict(value)
return value
def _ensure_state():
for key, value in STATE_DEFAULTS.items():
st.session_state.setdefault(key, _copy_default(value))
def _clear_run_state(*, keep_paper_input: bool = True) -> None:
"""Reset transient run UI state for a new pipeline/replay.
Never writes ``paper_input`` after the text-input widget exists. Preserves
``run_results_by_paper`` so other papers' results stay available.
"""
for key, value in STATE_DEFAULTS.items():
if key in WIDGET_STATE_KEYS or key in PRESERVED_STATE_KEYS:
continue
st.session_state[key] = _copy_default(value)
if not keep_paper_input:
# Defer clearing/replacing the widget value until before it is created.
st.session_state["_pending_paper_input"] = ""
def _store_paper_result(arxiv_id: str, result: dict) -> None:
if not arxiv_id:
return
cache = st.session_state.setdefault("run_results_by_paper", {})
cache[str(arxiv_id)] = dict(result)
def _paper_result_for_input(paper_input: str) -> Optional[dict]:
arxiv_id = replay_module.parse_arxiv_id(paper_input or "")
if not arxiv_id:
return None
cache = st.session_state.get("run_results_by_paper") or {}
entry = cache.get(arxiv_id)
return entry if isinstance(entry, dict) else None
def _apply_pending_paper_input() -> None:
"""Apply deferred paper_input changes before the text_input widget mounts."""
if "_pending_paper_input" not in st.session_state:
return
st.session_state["paper_input"] = st.session_state.pop("_pending_paper_input")
def _paper_1_url() -> str:
urls = list(EXAMPLES.values())
return urls[0] if urls else ""
def _set_demo_paper(url: str) -> None:
"""Pin the demo paper selection (source of truth across reruns)."""
st.session_state["live_run_mode"] = False
st.session_state["selected_demo_url"] = url
st.session_state["_pending_paper_input"] = url
def _on_pick_demo(url: str) -> None:
_set_demo_paper(url)
def _on_pick_live_run() -> None:
st.session_state["live_run_mode"] = True
st.session_state["_pending_paper_input"] = ""
def _select_paper_1() -> None:
"""Select demo Paper 1 in the picker (clears live-run mode)."""
_set_demo_paper(_paper_1_url())
def _set_system_method(method_id: str) -> None:
"""Pin System Run method and reset case when the method changes."""
prev = st.session_state.get("selected_system_method")
st.session_state["selected_system_method"] = method_id
if prev != method_id:
st.session_state["selected_system_case"] = system_run_module.default_case_for_method(
method_id
)
def _set_system_case(case_key: str) -> None:
st.session_state["selected_system_case"] = case_key
def _on_pick_system_method(method_id: str) -> None:
_set_system_method(method_id)
def _on_pick_system_case(case_key: str) -> None:
_set_system_case(case_key)
def _on_pick_system_prerun() -> None:
st.session_state["system_run_mode"] = "prerun"
def _on_pick_system_live() -> None:
st.session_state["system_run_mode"] = "live"
def _on_pick_live_method(method_id: str) -> None:
st.session_state["selected_live_method"] = method_id
def _empty_live_method_state() -> dict[str, Any]:
return {
"result": None,
"judged": None,
"logs": [],
"status": "Idle",
"pending_action": None, # "run" | "evaluate" | None
"ran_once": False,
"evaluated_once": False,
}
def _live_method_state(method_id: str) -> dict[str, Any]:
"""Return the mutable per-method Live Run slot (isolated across settings)."""
store = st.session_state.get("live_run_by_method")
if not isinstance(store, dict):
store = {}
st.session_state["live_run_by_method"] = store
slot = store.get(method_id)
if not isinstance(slot, dict):
slot = _empty_live_method_state()
store[method_id] = slot
else:
slot.setdefault("result", None)
slot.setdefault("judged", None)
slot.setdefault("logs", [])
slot.setdefault("status", "Idle")
slot.setdefault("pending_action", None)
slot.setdefault("ran_once", False)
slot.setdefault("evaluated_once", False)
return slot
def _render_system_live_run_view() -> None:
"""Live Run: one AVerImaTeC example + parametric / websearch_deep + Task A eval."""
try:
example = system_live_module.example_claim()
except Exception as exc:
st.error(f"Could not load demo claim: {exc}")
return
live_method = str(
st.session_state.get("selected_live_method") or "codeagent_parametric"
).strip()
method_ids = {m["id"] for m in system_live_module.METHOD_UI}
if live_method not in method_ids:
live_method = "codeagent_parametric"
st.session_state["selected_live_method"] = live_method
slot = _live_method_state(live_method)
status = str(slot.get("status") or "Idle")
busy = status in {"Running", "Evaluating"}
can_evaluate = bool(slot.get("result")) and not busy
run_label = "Re-run System" if slot.get("ran_once") else "Run SciFy System"
eval_label = "Re-evaluate" if slot.get("evaluated_once") else "Evaluate"
st.markdown('Example claim
', unsafe_allow_html=True)
paper_title = str(example.get("paper_title") or "").strip() or "Unknown paper"
st.markdown(
f"""
Claim
{_escape(_ensure_claim_period(example.get("claim")))}
Paper: {_escape(paper_title)}
""",
unsafe_allow_html=True,
)
st.markdown('Setting
', unsafe_allow_html=True)
method_cols = st.columns(len(system_live_module.METHOD_UI))
for i, item in enumerate(system_live_module.METHOD_UI):
with method_cols[i]:
st.button(
item["label"],
key=f"live_method::{item['id']}",
type="primary" if item["id"] == live_method else "secondary",
use_container_width=True,
on_click=_on_pick_live_method,
args=(item["id"],),
disabled=busy,
)
st.markdown('Action
', unsafe_allow_html=True)
run_col, eval_col = st.columns(2)
with run_col:
run_clicked = st.button(
run_label,
type="secondary",
use_container_width=True,
key=f"live_run_system_btn::{live_method}",
disabled=busy,
)
with eval_col:
eval_clicked = st.button(
eval_label,
type="secondary",
use_container_width=True,
key=f"live_eval_btn::{live_method}",
disabled=busy or not can_evaluate,
)
with st.container(key="live_run_terminal"):
log_box = st.empty()
logs = list(slot.get("logs") or [])
def _flush_logs() -> None:
body = "\n".join(logs[-400:]) if logs else "Waiting for run output…"
log_box.code(body, language="text")
st.caption(f"Status: {status}")
_flush_logs()
# Phase 1: click → mark this method busy and rerun so buttons grey out immediately.
if run_clicked and not busy:
slot["pending_action"] = "run"
slot["judged"] = None
slot["result"] = None
slot["logs"] = []
slot["status"] = "Running"
st.rerun()
if eval_clicked and not busy and slot.get("result"):
slot["pending_action"] = "evaluate"
slot["status"] = "Evaluating"
st.rerun()
# Phase 2: execute pending action while buttons stay disabled.
pending = slot.get("pending_action")
if pending == "run" and status == "Running":
slot["pending_action"] = None
logs = []
status = "Running"
def _log(line: str) -> None:
logs.append(line)
slot["logs"] = list(logs)
log_box.code("\n".join(logs[-400:]), language="text")
try:
result = system_live_module.run_live_method(live_method, log_callback=_log)
slot["result"] = result
slot["status"] = "Completed"
slot["ran_once"] = True
status = "Completed"
_log(f"Done. ingredients={len(result.get('ingredients') or [])}")
except Exception as exc:
slot["result"] = None
slot["status"] = "Failed"
slot["ran_once"] = True
status = "Failed"
_log(f"ERROR: {exc}")
st.rerun()
if pending == "evaluate" and status == "Evaluating" and slot.get("result"):
slot["pending_action"] = None
logs = list(slot.get("logs") or [])
status = "Evaluating"
def _log_eval(line: str) -> None:
if line.strip().startswith("{") and '"ok"' in line:
return
logs.append(line)
slot["logs"] = list(logs)
log_box.code("\n".join(logs[-400:]), language="text")
try:
judged = system_live_module.evaluate_live_run(
slot["result"],
log_callback=_log_eval,
)
slot["judged"] = judged
slot["status"] = "Evaluated"
slot["evaluated_once"] = True
status = "Evaluated"
_log_eval(
f"Evaluated. F1={float(judged.get('f1') or 0):.2f} "
f"R={float(judged.get('recall') or 0):.2f} "
f"P={float(judged.get('precision') or 0):.2f}"
)
except Exception as exc:
slot["status"] = "Eval failed"
slot["evaluated_once"] = True
status = "Eval failed"
_log_eval(f"ERROR: {exc}")
st.rerun()
run_result = slot.get("result")
judged = slot.get("judged")
if run_result:
case = system_live_module.live_result_to_case_card(run_result, judged)
# Claim/paper already shown above in the example block; card is results only.
result_html = _system_case_card_html(case)
st.markdown(
f'{result_html}
',
unsafe_allow_html=True,
)
def _normalize_abstract(text: str) -> str:
cleaned = re.sub(r"\s+", " ", (text or "").strip())
cleaned = re.sub(r"^(abstract)\s*[:.]?\s*", "", cleaned, flags=re.IGNORECASE)
# Drop trailing PDF footnote markers like ").1" / ".”2".
cleaned = re.sub(r"(?<=[.!?\)\]\"'”’])\d+$", "", cleaned)
return cleaned.strip()
def _metadata_from_paper_dir(paper_dir: Path) -> Optional[dict[str, str]]:
data = _load_json(paper_dir / "paper_metadata.json")
record = None
if isinstance(data, list) and data:
record = data[0] if isinstance(data[0], dict) else None
elif isinstance(data, dict):
record = data
if not record:
return None
title = str(record.get("title") or "").strip()
abstract = _normalize_abstract(str(record.get("abstract") or ""))
if not title and not abstract:
return None
return {"title": title, "abstract": abstract}
@st.cache_data(show_spinner=False, ttl=6 * 60 * 60)
def _fetch_arxiv_title_abstract(arxiv_id: str) -> Optional[dict[str, str]]:
url = f"http://export.arxiv.org/api/query?id_list={arxiv_id}"
try:
with urllib.request.urlopen(url, timeout=12) as resp: # noqa: S310
payload = resp.read()
except (urllib.error.URLError, TimeoutError, OSError):
return None
try:
root = ET.fromstring(payload)
except ET.ParseError:
return None
ns = {"atom": "http://www.w3.org/2005/Atom"}
entry = root.find("atom:entry", ns)
if entry is None:
return None
title = re.sub(r"\s+", " ", (entry.findtext("atom:title", default="", namespaces=ns) or "").strip())
abstract = _normalize_abstract(entry.findtext("atom:summary", default="", namespaces=ns) or "")
if not title and not abstract:
return None
return {"title": title, "abstract": abstract}
def _resolve_paper_preview(paper_input: str) -> Optional[dict[str, str]]:
"""Title/abstract for the selected paper (trace metadata, run dir, or arXiv)."""
arxiv_id = replay_module.parse_arxiv_id(paper_input or "")
if not arxiv_id:
return None
entry = _paper_result_for_input(paper_input)
if entry and entry.get("paper_dir_path"):
meta = _metadata_from_paper_dir(Path(entry["paper_dir_path"]))
if meta:
return meta
trace_dir = replay_module.find_trace_dir(paper_input)
if trace_dir is not None:
meta = _metadata_from_paper_dir(trace_dir / "processed_papers" / arxiv_id)
if meta:
return meta
return _fetch_arxiv_title_abstract(arxiv_id)
def _render_paper_preview(paper_input: str) -> None:
preview = _resolve_paper_preview(paper_input)
if not preview:
st.caption("Paper metadata unavailable for this ID yet.")
return
title = preview.get("title") or "Untitled paper"
abstract = preview.get("abstract") or ""
# Title as HTML; abstract as markdown so `$...$` goes through Streamlit KaTeX.
body = (
f""
)
if abstract:
body = f"{body}\n\n{abstract}"
st.markdown(body, unsafe_allow_html=True)
def _derive_work_step_states(events: list[str]) -> dict[int, str]:
"""Map pipeline step numbers to pending | active | done | failed."""
order = [n for n, _ in WORK_PANEL_STEPS]
started: set[int] = set()
completed: set[int] = set()
failed = False
pipeline_complete = False
for raw in events or []:
text = str(raw)
lower = text.lower()
if text == "Pipeline completed successfully.":
pipeline_complete = True
if "failed" in lower:
failed = True
m_done = re.search(r"Step\s+(\d+)\s+complete", text, re.IGNORECASE)
if m_done:
completed.add(int(m_done.group(1)))
continue
m_start = re.search(r"Step\s+(\d+)\s*/", text)
if m_start:
started.add(int(m_start.group(1)))
continue
if "[annotation]" in lower or "annotate target contributions" in lower:
if "complete" in lower or "skipped" in lower:
completed.add(8)
else:
started.add(8)
max_started = max(started) if started else 0
states: dict[int, str] = {}
for n in order:
if n in completed or (n in started and n < max_started):
states[n] = "done"
elif n in started:
states[n] = "failed" if failed else "active"
else:
states[n] = "pending"
if pipeline_complete and not failed:
for n in order:
# Replay traces often omit explicit completes for every step.
if n < 8 or 8 in completed or 8 in started:
states[n] = "done"
if 8 in completed:
states[8] = "done"
return states
def _display_work_status(status: str) -> str:
"""Steps badge only shows Idle / Running (Stopped/Failed/Completed → Idle)."""
text = (status or "").strip()
if not text:
return "Idle"
lower = text.lower()
if lower in {"running", "starting"} or text.startswith("Step "):
return "Running"
return "Idle"
def _work_panel_html(
events: list[str],
*,
status: str = "Idle",
mode_note: str = "",
) -> str:
states = _derive_work_step_states(events)
status_label = f"{_display_work_status(status)}{mode_note}"
rows: list[str] = []
for n, label in WORK_PANEL_STEPS:
state = states.get(n, "pending")
busy = " aria-busy='true'" if state == "active" else ""
rows.append(
""
""
f"{_esc(label)}"
""
)
body = (
f""
if rows
else "Waiting for the first step…
"
)
return (
""
"
"
"
Steps
"
f"
{_esc(status_label)}
"
"
"
f"{body}"
"
"
f"
{_esc(label)}
"
f"
{_esc(value)}
"
f"
{_esc(' · '.join(meta))}
" if meta else ""
return (
""
f"
{_esc(label)}
"
f"
{_esc(title)}
"
f"{meta_html}"
"
No grounded studies listed for this target contribution.
", unsafe_allow_html=True)
return
for item in studies:
title = item.get("ref_title") or item.get("title") or item.get("ref_id") or item.get("paper_id") or "Untitled reference"
meta = []
if item.get("_grounding_label"):
meta.append(str(item.get("_grounding_label")))
if item.get("role"):
meta.append(str(item.get("role")))
if item.get("ref_year"):
meta.append(str(item.get("ref_year")))
class_name = "cluster-card additional-study"
body = [f"{_esc(title)}
"]
if meta:
body.append(f"
{_esc(' · '.join(meta))}
")
if item.get("contribution"):
body.append(f"
Contribution. {_esc(item.get('contribution'))}
")
if item.get("rationale"):
body.append(f"
Rationale. {_esc(item.get('rationale'))}
")
body.append("
No annotation payload is available yet.
", unsafe_allow_html=True)
return
claims = payload.get("claims") or []
if not claims:
st.markdown("The run completed, but no target contributions were produced.
", unsafe_allow_html=True)
return
st.markdown(
"This decomposition is done in hindsight: SciPaths uses observed downstream citation clusters to identify reusable target contributions, then reconstructs the enabling contributions and groundings behind each one.
",
unsafe_allow_html=True,
)
for idx, claim in enumerate(claims, start=1):
claim_id = claim.get("claim_id") or f"C{idx}"
claim_text = claim.get("rewritten_claim") or claim.get("text") or "(missing target contribution text)"
ingredients = claim.get("ingredients") or []
discoveries = claim.get("enabling_discoveries") or []
grounded_studies = _collect_grounded_studies(discoveries, ingredients)
meta_bits = []
if claim.get("decision"):
meta_bits.append(str(claim.get("decision")))
if claim.get("cluster_id"):
meta_bits.append(f"cluster {claim.get('cluster_id')}")
meta_bits.append(f"{len(ingredients)} enabling contribution{'s' if len(ingredients) != 1 else ''}")
meta_bits.append(f"{len(grounded_studies)} grounded stud{'ies' if len(grounded_studies) != 1 else 'y'}")
st.markdown(
f"""
Target contribution {idx} · {_esc(claim_id)}
{_esc(claim_text)}
{_esc(' · '.join(meta_bits))}
""",
unsafe_allow_html=True,
)
left, right = st.columns([1.7, 1.0], gap="large")
with left:
st.markdown("Decomposition
", unsafe_allow_html=True)
if not ingredients:
st.markdown("No enabling contributions for this target contribution.
", unsafe_allow_html=True)
for ingredient_idx, ingredient in enumerate(ingredients, start=1):
annotation = ingredient.get("canonical_annotation") or {}
role = annotation.get("role") or ", ".join(annotation.get("roles") or []) or "UNSPECIFIED"
canonical_grounding = ingredient.get("canonical_grounding") or {}
extras = ingredient.get("additional_groundings") or []
grounding_parts = []
if canonical_grounding:
grounding_parts.append(
_grounding_html(canonical_grounding, "Grounding", "grounding")
)
for ref in extras:
if not isinstance(ref, dict):
continue
if canonical_grounding and (
ref.get("paper_id") == canonical_grounding.get("paper_id")
or ref.get("ref_id") == canonical_grounding.get("ref_id")
):
continue
grounding_parts.append(
_grounding_html(ref, "Grounding", "grounding")
)
if not grounding_parts:
canonical_ref_id = ingredient.get("canonical_ref_id") or "__NONE__"
grounding_parts.append(
""
"
Grounding
"
f"
{_esc(canonical_ref_id)}
"
"
"
f"
Groundings for enabling contribution {ingredient_idx}
"
+ "".join(grounding_parts)
+ "
{ingredient_idx}. {_esc(ingredient.get('ingredient') or '(missing enabling contribution)')}
{_esc(role)}
Contribution. {_esc(annotation.get('contribution') or '')}
Rationale. {_esc(annotation.get('rationale') or '')}
Evidence. {_esc(annotation.get('evidence_span') or '')}
{grounding_block}
""",
unsafe_allow_html=True,
)
with right:
st.markdown("Groundings
", unsafe_allow_html=True)
_render_reference_list(discoveries, ingredients)
def _render_clusters_tab(discovery: Optional[dict], contributions: list[dict]):
if not discovery:
st.markdown("No refined cluster file is available yet.
", unsafe_allow_html=True)
return
st.markdown(
"Citation clusters look forward: they summarize how later papers use or extend the input paper, showing which contributions became reusable.
",
unsafe_allow_html=True,
)
clusters = discovery.get("clusters") or []
dropped = discovery.get("dropped_clusters") or []
if not clusters:
st.markdown("No valid downstream usage clusters survived refinement and filtering.
", unsafe_allow_html=True)
if dropped:
with st.expander(f"Dropped clusters ({len(dropped)})", expanded=False):
st.json(dropped)
return
for cluster in clusters:
cluster_id = cluster.get("cluster_id", "")
rep = cluster.get("representative_claim") or cluster.get("cluster_title") or "(missing representative claim)"
count = _safe_int(cluster.get("count"), len(cluster.get("claim_indices") or []))
source_ids = cluster.get("source_cluster_ids") or []
merge_rationale = cluster.get("merge_rationale") or ""
st.markdown(
f"""
{_esc(rep)}
Cluster {_esc(cluster_id)} · {count} contribution instance{'s' if count != 1 else ''}
""",
unsafe_allow_html=True,
)
meta_cols = st.columns([1.3, 1.3, 1.4])
with meta_cols[0]:
st.caption("Cluster ID")
st.code(str(cluster_id), language="text")
with meta_cols[1]:
st.caption("Source clusters")
st.code(", ".join(str(x) for x in source_ids) if source_ids else "singleton", language="text")
with meta_cols[2]:
st.caption("Merge rationale")
st.write(merge_rationale or "—")
claim_indices = cluster.get("claim_indices") or []
if claim_indices:
with st.expander(f"Linked contribution instances ({len(claim_indices)})", expanded=False):
for idx in claim_indices:
try:
j = int(idx)
except Exception:
continue
if 0 <= j < len(contributions):
item = contributions[j] or {}
title = item.get("citing_title") or item.get("citing_paper_id") or "Unknown citing paper"
claim = item.get("paper_claim") or item.get("claim") or "(missing claim)"
rationale = item.get("rationale") or ""
evidence = item.get("evidence_span") or ""
st.markdown(f"**{title}**")
st.write(claim)
if rationale:
st.caption(f"Rationale: {rationale}")
if evidence:
st.caption(f"Evidence: {evidence}")
st.divider()
if dropped:
with st.expander(f"Dropped clusters ({len(dropped)})", expanded=False):
st.json(dropped)
def run_replay_stream(
paper_input: str,
work_panel=None,
metrics_slot=None,
graph_slot=None,
courier_slot=None,
) -> bool:
"""Replay a saved demo trace into session state. Returns False if no trace exists."""
trace = replay_module.load_trace(paper_input)
if not trace:
return False
_clear_run_state(keep_paper_input=True)
st.session_state["run_status"] = "Running"
st.session_state["replay_mode"] = True
# Graph can read the finished trace artifacts while steps reveal progressively.
st.session_state["paper_dir_path"] = trace.get("paper_dir_path")
st.session_state["annotation_payload_path"] = trace.get("annotation_payload_path")
try:
import neo4j_workflow as _neo
_neo.reset_run(_workflow_run_id(paper_input))
except Exception:
pass
panel = work_panel if work_panel is not None else st.empty()
events: list[str] = []
_render_work_panel(panel, events, status="Running")
_push_graph(courier_slot, paper_input, events)
for line in trace.get("run_events") or []:
display_line = _display_log_line(str(line))
if not display_line:
continue
if display_line not in events:
events.append(display_line)
_render_work_panel(panel, events, status="Running")
_push_graph(courier_slot, paper_input, events)
delay = REPLAY_STEP_DELAY_SEC
if re.search(r"Step\s+\d+\s+complete", display_line, re.IGNORECASE):
delay += REPLAY_STEP_HOLD_SEC
time.sleep(max(0.0, delay))
# Commit final replay payload for the Clusters / Decomposition tabs.
st.session_state["run_status"] = trace.get("run_status") or trace.get("status") or "Completed"
st.session_state["run_logs"] = list(trace.get("run_logs") or events)
st.session_state["run_events"] = list(trace.get("run_events") or events)
st.session_state["artifact_path"] = trace.get("artifact_path")
st.session_state["run_dir_path"] = trace.get("run_dir_path")
st.session_state["paper_dir_path"] = trace.get("paper_dir_path")
st.session_state["annotation_payload_path"] = trace.get("annotation_payload_path")
st.session_state["annotation_skipped_reason"] = trace.get("annotation_skipped_reason")
st.session_state["pipeline_stopped_reason"] = trace.get("pipeline_stopped_reason")
st.session_state["pipeline_failed_reason"] = trace.get("pipeline_failed_reason")
st.session_state["remote_artifact_ref"] = ""
st.session_state["replay_mode"] = True
arxiv_id = str(trace.get("arxiv_id") or replay_module.parse_arxiv_id(paper_input) or "")
_store_paper_result(
arxiv_id,
{
"run_status": st.session_state["run_status"],
"run_logs": list(st.session_state["run_logs"]),
"run_events": list(st.session_state["run_events"]),
"artifact_path": st.session_state.get("artifact_path"),
"run_dir_path": st.session_state.get("run_dir_path"),
"paper_dir_path": st.session_state.get("paper_dir_path"),
"annotation_payload_path": st.session_state.get("annotation_payload_path"),
"annotation_skipped_reason": st.session_state.get("annotation_skipped_reason"),
"pipeline_stopped_reason": st.session_state.get("pipeline_stopped_reason"),
"pipeline_failed_reason": st.session_state.get("pipeline_failed_reason"),
"replay_mode": True,
},
)
_render_work_panel(
panel,
st.session_state["run_events"],
status=st.session_state["run_status"],
)
_push_graph(courier_slot, paper_input, st.session_state["run_events"])
_refresh_overview_for_paper(paper_input, metrics_slot)
return True
def run_two_pass_annotation(
paper_dir: Path,
annotation_output_root: Path,
llm_provider: str,
llm_model: str,
formatter_model: str,
judge_model: str,
candidate_count: int,
):
paper = load_paper_package(paper_dir)
pipeline = TwoPassAnnotationPipeline(
provider=llm_provider,
model=llm_model,
formatter_model=formatter_model or None,
judge_model=judge_model or None,
output_root=annotation_output_root,
annotator_id="streamlit_hf_space",
candidate_count=max(1, int(candidate_count)),
formatter_max_attempts=3,
include_reference_examples=True,
prompt_profile="full",
)
result = pipeline.run(paper)
return result.result, result.run_dir
def run_pipeline_stream(
paper_input: str,
source_root: str,
output_root: str,
llm_provider: str,
llm_model: str,
llm_model_step4: str,
formatter_model: str,
judge_model: str,
candidate_count: int,
work_panel=None,
metrics_slot=None,
graph_slot=None,
courier_slot=None,
):
gemini_key = get_secret("GEMINI_API_KEY")
if gemini_key:
os.environ["GEMINI_API_KEY"] = gemini_key
# Drop previous success/failure UI before this run starts rendering progress.
_clear_run_state(keep_paper_input=True)
st.session_state["run_status"] = "Starting"
try:
import neo4j_workflow as _neo
_neo.reset_run(_workflow_run_id(paper_input))
except Exception:
pass
cfg = PipelineConfig(
repo_root=REPO_ROOT,
source_root=Path(source_root).expanduser().resolve(),
paper_input=paper_input.strip(),
llm_provider=llm_provider.strip() or "gemini",
llm_model=llm_model.strip() or "gemini-3.1-pro-preview",
llm_model_step4=llm_model_step4.strip() or "gemini-3-flash-preview",
model_path="Deep-Citation/Workspace/acl_scicite_wksp_trl/best_model.pt",
model_data_dir="Deep-Citation/Data",
model_class_def="Deep-Citation/Data/class_def.json",
model_lm="scibert",
device="cpu",
embedding_model="sentence-transformers/all-mpnet-base-v2",
)
panel = work_panel if work_panel is not None else st.empty()
status = "Starting"
logs: list[str] = []
events: list[str] = []
seen_events: set[str] = set()
artifact_path = None
annotation_payload_path = None
annotation_skipped_reason = None
run_summary = None
pipeline_stopped_reason = None
pipeline_failed_reason = None
def render_activity(items: list[str], *, current_status: str | None = None):
_render_work_panel(
panel,
items,
status=current_status or status,
mode_note="",
)
_push_graph(courier_slot, paper_input, items)
def append_display_line(line: str):
nonlocal status
display_line = _display_log_line(line)
if not display_line:
return
logs.append(display_line)
status = _status_from_line(display_line, status)
event = _format_step_event(display_line)
if event and event not in seen_events:
seen_events.add(event)
events.append(event)
render_activity(events, current_status=status)
render_activity(events, current_status=status)
for line, maybe_artifact in runner_module.run_pipeline(cfg, Path(output_root).expanduser().resolve()):
if line:
if line.strip() == "Pipeline completed successfully.":
if maybe_artifact:
artifact_path = maybe_artifact
continue
display_line = _display_log_line(line)
if display_line:
logs.append(display_line)
status = _status_from_line(display_line, status)
if display_line.startswith("Pipeline stopped:"):
pipeline_stopped_reason = display_line
if "failed" in display_line.lower():
pipeline_failed_reason = display_line
event = _format_step_event(display_line)
if event and event not in seen_events:
seen_events.add(event)
events.append(event)
if maybe_artifact:
artifact_path = maybe_artifact
try:
job_dir = Path(str(maybe_artifact)).with_suffix("")
paper_id = runner_module.parse_arxiv_id(paper_input.strip())
st.session_state["paper_dir_path"] = str(job_dir / "processed_papers" / paper_id)
except Exception:
pass
render_activity(events, current_status=status)
run_dir_path = None
paper_dir_path = None
remote_artifact_ref = ""
if artifact_path:
job_dir = Path(str(artifact_path)).with_suffix("")
run_dir_path = str(job_dir)
paper_id = runner_module.parse_arxiv_id(paper_input.strip())
paper_dir = job_dir / "processed_papers" / paper_id
paper_dir_path = str(paper_dir)
if pipeline_failed_reason:
annotation_skipped_reason = f"{pipeline_failed_reason} Annotation was not run."
elif pipeline_stopped_reason:
annotation_skipped_reason = f"{pipeline_stopped_reason} Annotation was not run."
else:
discovery = _load_json(paper_dir / "usage_discovery_from_contributions.json") or {}
refined_clusters = discovery.get("clusters") or []
if not refined_clusters:
annotation_skipped_reason = "No valid downstream usage clusters remained after refinement and filtering. Annotation was skipped."
logs.append("[annotation] skipped: no refined downstream usage clusters")
else:
append_display_line("[annotation] starting cluster-first two-pass annotation")
try:
run_output, annotation_run_dir = run_two_pass_annotation(
paper_dir=paper_dir,
annotation_output_root=job_dir / "two_pass_outputs",
llm_provider=llm_provider,
llm_model=llm_model,
formatter_model=formatter_model,
judge_model=judge_model,
candidate_count=candidate_count,
)
payload_path = run_output.get("ui_payload_path") if isinstance(run_output, dict) else None
if payload_path and Path(payload_path).exists():
annotation_payload_path = str(Path(payload_path))
append_display_line(f"[annotation] complete: {annotation_run_dir}")
except Exception as exc:
pipeline_failed_reason = f"Annotation failed: {exc}"
annotation_skipped_reason = pipeline_failed_reason
logs.append(f"[annotation] failed: {exc}")
logs.append("[upload] uploading run artifact to Hugging Face dataset")
remote_artifact_ref = upload_run_artifact(job_dir)
if remote_artifact_ref:
logs.append(f"[upload] {remote_artifact_ref}")
else:
logs.append("[upload] skipped: RUNS_REPO_ID/HF_WRITE_TOKEN not configured")
if not pipeline_stopped_reason and not pipeline_failed_reason:
append_display_line("Pipeline completed successfully.")
if pipeline_failed_reason:
status = "Failed"
elif artifact_path and pipeline_stopped_reason:
status = "Stopped"
else:
status = "Completed" if artifact_path else "Failed"
render_activity(events, current_status=status)
st.session_state["run_status"] = status
st.session_state["run_logs"] = logs
st.session_state["run_events"] = events
st.session_state["artifact_path"] = artifact_path
st.session_state["run_dir_path"] = run_dir_path
st.session_state["paper_dir_path"] = paper_dir_path
st.session_state["annotation_payload_path"] = annotation_payload_path
st.session_state["annotation_skipped_reason"] = annotation_skipped_reason
st.session_state["pipeline_stopped_reason"] = pipeline_stopped_reason
st.session_state["pipeline_failed_reason"] = pipeline_failed_reason
st.session_state["run_summary"] = run_summary
st.session_state["remote_artifact_ref"] = remote_artifact_ref
st.session_state["replay_mode"] = False
arxiv_id = replay_module.parse_arxiv_id(paper_input) or ""
_store_paper_result(
arxiv_id,
{
"run_status": status,
"run_logs": list(logs),
"run_events": list(events),
"artifact_path": artifact_path,
"run_dir_path": run_dir_path,
"paper_dir_path": paper_dir_path,
"annotation_payload_path": annotation_payload_path,
"annotation_skipped_reason": annotation_skipped_reason,
"pipeline_stopped_reason": pipeline_stopped_reason,
"pipeline_failed_reason": pipeline_failed_reason,
"replay_mode": False,
},
)
_push_graph(courier_slot, paper_input, events)
_refresh_overview_for_paper(paper_input, metrics_slot)
def _load_result_bundle(paper_input: Optional[str] = None):
"""Load clusters/decomposition for the selected paper only (session cache)."""
selected = paper_input if paper_input is not None else st.session_state.get("paper_input", "")
entry = _paper_result_for_input(selected)
if not entry:
return None, None, [], None
paper_dir_path = entry.get("paper_dir_path")
annotation_payload_path = entry.get("annotation_payload_path")
paper_dir = Path(paper_dir_path) if paper_dir_path else None
payload = _load_json(Path(annotation_payload_path)) if annotation_payload_path else None
discovery = (
_load_json(paper_dir / "usage_discovery_from_contributions.json")
if paper_dir and paper_dir.exists()
else None
)
contributions_data = (
_load_json(paper_dir / "usage_contributions.json")
if paper_dir and paper_dir.exists()
else None
)
contributions = (contributions_data or {}).get("contributions") or []
return paper_dir, discovery, contributions, payload
def _overview_counts(
payload: Optional[dict], discovery: Optional[dict]
) -> tuple[int, int, int, int]:
claims = (payload or {}).get("claims") or []
ingredients = sum(len(claim.get("ingredients") or []) for claim in claims)
studies = sum(
len(_collect_grounded_studies(claim.get("enabling_discoveries") or [], claim.get("ingredients") or []))
for claim in claims
)
clusters = len((discovery or {}).get("clusters") or [])
return clusters, len(claims), ingredients, studies
def _overview_html(payload: Optional[dict], discovery: Optional[dict]) -> str:
clusters, n_claims, ingredients, studies = _overview_counts(payload, discovery)
return (
""
f"{_metric_card('Refined clusters', clusters)}"
f"{_metric_card('Target contributions', n_claims)}"
f"{_metric_card('Enabling contributions', ingredients)}"
f"{_metric_card('Grounded studies', studies)}"
"
"
)
def _render_overview(
payload: Optional[dict],
discovery: Optional[dict],
placeholder=None,
) -> None:
html_body = _overview_html(payload, discovery)
if placeholder is not None:
placeholder.markdown(html_body, unsafe_allow_html=True)
else:
st.markdown(html_body, unsafe_allow_html=True)
def _refresh_overview_for_paper(paper_input: str, placeholder) -> None:
"""Update metrics in-place after a run finishes (labels stay put; counts change)."""
if placeholder is None:
return
_paper_dir, discovery, _contributions, payload = _load_result_bundle(paper_input)
_render_overview(payload, discovery, placeholder=placeholder)
def _build_public_export(discovery: Optional[dict], payload: Optional[dict]) -> dict:
claims = []
for claim in (payload or {}).get("claims") or []:
if not isinstance(claim, dict):
continue
ingredients = []
for ingredient in claim.get("ingredients") or []:
if not isinstance(ingredient, dict):
continue
ingredients.append({
"ingredient_id": ingredient.get("ingredient_id"),
"enabling_contribution": ingredient.get("ingredient"),
"canonical_annotation": ingredient.get("canonical_annotation") or {},
"primary_grounding": ingredient.get("canonical_grounding") or {},
"additional_groundings": ingredient.get("additional_groundings") or [],
})
claims.append({
"claim_id": claim.get("claim_id"),
"target_contribution": claim.get("rewritten_claim") or claim.get("text"),
"cluster_id": claim.get("cluster_id"),
"decision": claim.get("decision"),
"enabling_contributions": ingredients,
"grounded_studies": _collect_grounded_studies(claim.get("enabling_discoveries") or [], claim.get("ingredients") or []),
})
return {
"citation_clusters": (discovery or {}).get("clusters") or [],
"target_contribution_decompositions": claims,
}
def _escape(text: Any) -> str:
return html.escape(str(text or ""), quote=True)
def _format_analysis_bullet(text: Any) -> str:
"""Bold leading taxonomy tag; render as 'Tag:' instead of '[Tag]'."""
raw = str(text or "")
match = re.match(r"^\[([^\]]+)\]\s*(.*)$", raw, flags=re.DOTALL)
if not match:
return _escape(raw)
tag = match.group(1).strip()
rest = match.group(2)
label = f"{tag}:"
if rest:
return f"{_escape(label)} {_escape(rest)}"
return f"{_escape(label)}"
def _render_gold_item(
item: dict, *, paired: bool = False, show_match_notes: bool = True, show_judge: bool = True
) -> str:
badge = str(item.get("badge") or "miss")
role = _escape(item.get("role") or "")
desc = _escape(item.get("description") or "")
notes = []
if show_match_notes:
for note in item.get("match_notes") or []:
cls = "system-match-note partial" if badge == "partial" else "system-match-note"
notes.append(f"{_escape(note)}
")
if show_judge:
judge = item.get("judge_note") or ""
if judge:
notes.append(f"{_escape(judge)}
")
paired_cls = " is-paired" if paired else ""
return (
f""
"
"
f"{badge}"
f"{role}"
"
"
f"
{desc}
"
f"{''.join(notes)}"
"
{_escape(rationale)}
" if rationale else ""
paired_cls = " is-paired" if paired else ""
return (
f""
"
"
f"{_escape(idx)}"
f"{role}"
"
"
f"
{desc}
"
f"{rat_html}"
"
None
"
return f"Gold ingredients
{body}"
def _render_pred_column(items: list[dict]) -> str:
rows = [_render_pred_item(item) for item in items]
body = "".join(rows) if rows else "None
"
return f"SciFy Predictions
{body}"
def _render_paired_ingredient_grid(case: dict) -> str:
"""Success layout: gold left, predictions right, lines between matched pairs."""
gold = case.get("gold_ingredients") or []
predicted = case.get("predicted_ingredients") or []
rows = system_run_module.build_pair_rows(gold, predicted)
body = [
"",
"
Gold ingredients
SciFy Predictions
",
]
for row in rows:
linked = bool(row.get("linked"))
gold_item = row.get("gold")
pred_item = row.get("pred")
left = (
# Pair lines already show the match; keep judge/rationale text.
_render_gold_item(
gold_item, paired=linked, show_match_notes=False, show_judge=True
)
if gold_item
else "
"
)
right = (
_render_pred_item(pred_item, paired=linked, show_rationale=True)
if pred_item
else "
"
)
link_cls = "system-pair-link" if linked else "system-pair-link is-empty"
body.append(
f"
"
)
body.append("
{_escape(q)}" for q in (panel.get("queries") or [])
)
docs = "".join(
f"{_escape(d)}" for d in (panel.get("retrieved_docs") or [])
)
return (
""
)
if kind == "deep":
queries = "".join(
f"{_escape(q)}" for q in (panel.get("queries") or [])
)
hits = "".join(f"{_escape(h)}" for h in (panel.get("hits") or []))
deep_items = []
for item in panel.get("deep_fetches") or []:
cls = "ok" if item.get("ok") else "bad"
deep_items.append(
f"{_escape(item.get('url'))}"
f"{_escape(item.get('outcome'))}
"
)
return (
""
)
return ""
def _ensure_claim_period(text: Any) -> str:
claim = str(text or "").strip()
if claim and claim[-1] not in ".!?":
claim += "."
return claim
def _system_claim_header_html(case: dict) -> str:
"""Claim + paper outside the result card (Decomposition-style claim layout)."""
paper_title = str(case.get("paper_title") or "").strip()
paper_line = (
f'Paper: {_escape(paper_title)}
'
if paper_title
else ""
)
return (
''
'
'
'
Claim
'
f'
{_escape(_ensure_claim_period(case.get("claim")))}
'
f"{paper_line}"
"
"
"
"
"Unevaluated"
"
"
)
else:
pills = (
""
f"F1 {f1:.2f}"
f"Recall {recall:.2f}"
f"Precision {precision:.2f}"
"
"
)
# Success: gold | predictions with pair connector lines.
# Failure: keep the original two-column layout (pairing UI deferred).
# Featured / unevaluated live runs: predictions only.
if case_key == "success":
ingredients_html = _render_paired_ingredient_grid(case)
elif case_key == "featured":
pred_col = _render_pred_column(case.get("predicted_ingredients") or [])
ingredients_html = f'{pred_col}
'
else:
gold_col = _render_gold_column(case.get("gold_ingredients") or [])
pred_col = _render_pred_column(case.get("predicted_ingredients") or [])
ingredients_html = (
f'{gold_col}{pred_col}
'
)
tool_html = _render_tool_panel_html(case.get("tool_panel") or {})
return (
f''
f'
{_escape(case.get("case_chip"))}
'
f"{pills}"
f"{ingredients_html}"
f"{tool_html}"
"
"
"
No analysis available for this case.
"
"
",
]
for block in blocks:
btype = block.get("type")
if btype == "diagnosis":
text = str(block.get("text") or "")
diag_cls = "system-diagnosis is-part" if tone == "part" else "system-diagnosis"
if block.get("strong_first") and text.startswith("Diagnosis."):
rest = text[len("Diagnosis.") :].lstrip()
html_parts.append(
f"
Diagnosis. {_escape(rest)}
"
)
elif block.get("strong_first"):
lines = text.split("\n", 1)
if len(lines) == 2:
html_parts.append(
f"
{_escape(lines[0])}"
f"
{_escape(lines[1])}
"
)
else:
html_parts.append(
f"
{_escape(text)}
"
)
else:
html_parts.append(f"
{_escape(text)}
")
elif btype == "two_col":
cols = []
for col in block.get("cols") or []:
items = "".join(
f"
{_format_analysis_bullet(x)}" for x in (col.get("items") or [])
)
cols.append(
"
"
f"
{_escape(col.get('title'))}
"
f"
"
"
"
)
html_parts.append(f"
{''.join(cols)}
")
elif btype == "list":
items = "".join(
f"
{_format_analysis_bullet(x)}" for x in (block.get("items") or [])
)
html_parts.append(f"
{_escape(block.get('title'))}
")
html_parts.append("
SciFy System Run on SciPaths Claims
SciFy CodeAgent
""",
unsafe_allow_html=True,
)
st.markdown('Mode
', unsafe_allow_html=True)
mode_cols = st.columns(2)
with mode_cols[0]:
st.button(
"Pre-run Cases",
key="system_mode::prerun",
type="primary" if mode == "prerun" else "secondary",
use_container_width=True,
on_click=_on_pick_system_prerun,
)
with mode_cols[1]:
st.button(
"Live Run",
key="system_mode::live",
type="primary" if mode == "live" else "secondary",
use_container_width=True,
on_click=_on_pick_system_live,
)
if mode == "live":
_render_system_live_run_view()
return
st.markdown('Setting
', unsafe_allow_html=True)
method_items = system_run_module.SYSTEM_RUN_METHODS
method_cols = st.columns(len(method_items))
for i, item in enumerate(method_items):
with method_cols[i]:
mid = item["id"]
st.button(
item["label"],
key=f"system_method::{mid}",
type="primary" if mid == method_id else "secondary",
use_container_width=True,
on_click=_on_pick_system_method,
args=(mid,),
)
st.markdown('Case
', unsafe_allow_html=True)
case_cols = st.columns(max(len(available_cases), 1))
for i, (ckey, clabel) in enumerate(available_cases):
with case_cols[i]:
st.button(
clabel,
key=f"system_case::{method_id}::{ckey}",
type="primary" if ckey == case_key else "secondary",
use_container_width=True,
on_click=_on_pick_system_case,
args=(ckey,),
)
try:
case = system_run_module.get_case(method_id, case_key)
except Exception as exc:
st.error(f"Could not load case study: {exc}")
return
claim_html = _system_claim_header_html(case)
result_html = _system_case_card_html(case)
analysis_html = _system_analysis_panel_html(
method_id, case_key, tone=str(case.get("outcome_tone") or "bad")
)
st.markdown(claim_html, unsafe_allow_html=True)
result_col, analysis_col = st.columns([1.25, 1], gap="large")
with result_col:
st.markdown(
f'{result_html}
',
unsafe_allow_html=True,
)
with analysis_col:
st.markdown(analysis_html, unsafe_allow_html=True)
def _render_annotation_process_view(
*,
source_root: str,
output_root: str,
llm_provider: str,
llm_model: str,
llm_model_step4: str,
formatter_model: str,
judge_model: str,
candidate_count: int,
):
if not get_secret("GEMINI_API_KEY"):
st.warning("GEMINI_API_KEY is missing for live runs. Demo examples still work without it.")
# Pin demo selection so post-run reruns cannot jump to another paper.
if not st.session_state.get("live_run_mode"):
demo = str(st.session_state.get("selected_demo_url") or "").strip()
if not demo:
_select_paper_1()
else:
# Re-assert the locked demo before the picker / text input mounts.
st.session_state["_pending_paper_input"] = demo
_apply_pending_paper_input()
current_paper = st.session_state.get("paper_input", "")
selected_demo = str(st.session_state.get("selected_demo_url") or "").strip()
selected_demo_id = replay_module.parse_arxiv_id(selected_demo)
st.markdown(
"""
Annotation Process
Forecasting pathways to scientific discovery
""",
unsafe_allow_html=True,
)
live_run_mode = bool(st.session_state.get("live_run_mode"))
st.markdown('Paper
', unsafe_allow_html=True)
example_items = list(EXAMPLES.items())
picker_count = len(example_items) + 1 # demos + Live run
cols = st.columns(picker_count)
for i, (button_label, value) in enumerate(example_items):
with cols[i]:
example_id = replay_module.parse_arxiv_id(value)
is_selected = (not live_run_mode) and bool(
selected_demo_id and example_id == selected_demo_id
)
st.button(
button_label,
key=f"example::{button_label}",
type="primary" if is_selected else "secondary",
use_container_width=True,
on_click=_on_pick_demo,
args=(value,),
)
with cols[-1]:
st.button(
"Live Run",
key="example::live_run",
icon=":material/edit_note:",
type="primary" if live_run_mode else "secondary",
use_container_width=True,
on_click=_on_pick_live_run,
)
tabs = st.tabs(TAB_NAMES)
with tabs[0]:
if live_run_mode:
st.caption("Enter any arXiv paper and run the live pipeline.")
paper_input = st.text_input(
"arXiv URL or ID",
key="paper_input",
placeholder="https://arxiv.org/abs/2311.14919",
)
else:
paper_input = selected_demo or current_paper
if not paper_input.strip():
st.caption("Select a demo paper above, or choose Live Run to enter a custom arXiv ID.")
paper_result = _paper_result_for_input(paper_input) if paper_input.strip() else None
status = (paper_result or {}).get("run_status") or "Idle"
events = list((paper_result or {}).get("run_events") or [])
work_panel = None
graph_slot = None
courier_slot = None
run_clicked = False
if paper_input.strip():
left_col, right_col = st.columns([1.55, 1], gap="large")
with left_col:
_render_paper_preview(paper_input)
with right_col:
work_panel = st.empty()
_render_work_panel(work_panel, events, status=status)
run_clicked = st.button(
"Run pipeline",
type="primary",
use_container_width=True,
key="run_pipeline_btn",
)
# Persistent graph canvas (mounted once) + invisible courier that
# streams incremental model updates into it without re-emitting it.
graph_slot = st.empty()
courier_slot = st.empty()
_mount_graph_shell(graph_slot)
_push_graph(courier_slot, paper_input, events)
if paper_result and paper_result.get("pipeline_failed_reason"):
st.error(paper_result["pipeline_failed_reason"])
if paper_result and paper_result.get("annotation_skipped_reason"):
st.warning(paper_result["annotation_skipped_reason"])
if (
paper_result
and paper_result.get("pipeline_stopped_reason")
and not paper_result.get("pipeline_failed_reason")
):
st.warning(paper_result["pipeline_stopped_reason"])
# Stable slot so only metric values change when a run finishes (not a full remount flash).
metrics_slot = st.empty()
paper_dir, discovery, contributions, payload = _load_result_bundle(paper_input)
_render_overview(payload, discovery, placeholder=metrics_slot)
if run_clicked:
if not paper_input.strip():
_clear_run_state()
st.session_state["pipeline_failed_reason"] = "Paper input is required."
st.session_state["run_status"] = "Failed"
st.rerun()
# Demo papers use saved traces; Live run always executes the live pipeline.
used_trace = (
False
if live_run_mode
else run_replay_stream(
paper_input,
work_panel=work_panel,
metrics_slot=metrics_slot,
graph_slot=graph_slot,
courier_slot=courier_slot,
)
)
if not used_trace:
_clear_run_state()
st.session_state["run_status"] = "Starting"
st.session_state["replay_mode"] = False
run_pipeline_stream(
paper_input=paper_input,
source_root=source_root,
output_root=output_root,
llm_provider=llm_provider,
llm_model=llm_model,
llm_model_step4=llm_model_step4,
formatter_model=formatter_model,
judge_model=judge_model,
candidate_count=candidate_count,
work_panel=work_panel,
metrics_slot=metrics_slot,
graph_slot=graph_slot,
courier_slot=courier_slot,
)
# Keep the same demo selected after finish; do not advance to Paper 2.
# No st.rerun() here: the stream already painted the final work panel,
# overview metrics, and graph via placeholders/courier. A full rerun
# remounts the page (and graph iframe) and looks like a refresh.
if not live_run_mode and paper_input.strip():
st.session_state["live_run_mode"] = False
st.session_state["selected_demo_url"] = paper_input.strip()
with tabs[1]:
paper_dir, discovery, contributions, payload = _load_result_bundle(
st.session_state.get("paper_input", "")
or st.session_state.get("selected_demo_url", "")
)
_render_clusters_tab(discovery, contributions)
with tabs[2]:
paper_dir, discovery, contributions, payload = _load_result_bundle(
st.session_state.get("paper_input", "")
or st.session_state.get("selected_demo_url", "")
)
_render_claims_tab(payload)
def main():
llm_provider = os.getenv("LLM_PROVIDER", "gemini")
llm_model = os.getenv("LLM_MODEL", "gemini-3.1-pro-preview")
llm_model_step4 = os.getenv("LLM_MODEL_STEP4", "gemini-3-flash-preview")
formatter_model = os.getenv("ANNOTATION_FORMATTER_MODEL", "gemini/gemini-3.1-pro-preview")
judge_model = os.getenv("ANNOTATION_JUDGE_MODEL", "gemini/gemini-3.1-pro-preview")
candidate_count = int(os.getenv("ANNOTATION_CANDIDATE_COUNT", "3"))
source_root = DEFAULT_SOURCE_ROOT
output_root = DEFAULT_OUTPUT_ROOT
st.set_page_config(
page_title="SciPaths",
page_icon="🔬",
layout="wide",
initial_sidebar_state="collapsed",
)
st.markdown(CUSTOM_CSS, unsafe_allow_html=True)
_ensure_state()
if st.button("Reset session", type="tertiary", key="reset_session_btn"):
_reset_all_session_state()
st.rerun()
nav_col, main_col = st.columns([1, 5.5], gap="medium")
with nav_col:
current = st.session_state.get("app_view", DEFAULT_APP_VIEW)
if current not in APP_VIEWS:
current = DEFAULT_APP_VIEW
st.session_state["app_view"] = current
st.markdown(
"""
""",
unsafe_allow_html=True,
)
for item in APP_NAV:
view = item["view"]
icon = item.get("icon") or ""
is_active = view == current
if st.button(
view,
key=f"nav::{view}",
icon=icon,
type="primary" if is_active else "secondary",
use_container_width=True,
):
if view != current:
st.session_state["app_view"] = view
if view == "Annotation Process":
_select_paper_1()
st.rerun()
selected = st.session_state.get("app_view", DEFAULT_APP_VIEW)
with main_col:
if selected == "System Run":
_render_system_run_view()
else:
_render_annotation_process_view(
source_root=source_root,
output_root=output_root,
llm_provider=llm_provider,
llm_model=llm_model,
llm_model_step4=llm_model_step4,
formatter_model=formatter_model,
judge_model=judge_model,
candidate_count=candidate_count,
)
if __name__ == "__main__":
main()