diff --git a/src/aare/common/sample_history.py b/src/aare/common/sample_history.py
new file mode 100644
index 00000000..74cf1b65
--- /dev/null
+++ b/src/aare/common/sample_history.py
@@ -0,0 +1,153 @@
+"""Sample run-history derived from aareDB sample events.
+
+DRAFT. aareDB exposes ``Sample.events`` (``List[SampleEventResponse]`` with
+``event_type``/``comment``/``timestamp``) via ``get_samples_by_beamtime``. No DB
+schema change is needed: this module groups those events into *runs* (a run = a
+mount cycle) and maps each run's events to a per-stage outcome for the 6-LED view
+(Mount · Center · Raster · XRF · Collect · Process).
+
+Pure + dependency-light (events are duck-typed) so it can be unit-tested without
+aareDB and reused by both the server and the GUI.
+"""
+
+from __future__ import annotations
+
+from enum import Enum
+from typing import Optional
+
+from pydantic import BaseModel
+
+STAGES = ("Mount", "Center", "Raster", "XRF", "Collect", "Process")
+
+
+class StageStatus(str, Enum):
+    OK = "ok"
+    ERROR = "error"
+    WARN = "warn"
+    RUNNING = "running"
+    SKIP = "skip"          # not attempted in this run
+    UNKNOWN = "unknown"    # no information
+
+
+# stage -> (success event names, failure event names, in-progress event names)
+_STAGE_EVENTS: dict[str, tuple[set, set, set]] = {
+    "Mount":   ({"Mounted"}, {"MountFailed", "Lost"}, {"Mounting"}),
+    "Center":  ({"Centered", "LoopFaceDetected"},
+                {"ALCFailed", "AXCFailed", "LoopFaceDetectFailed"},
+                {"Centering", "LoopFaceDetecting"}),
+    "Raster":  ({"Rastered"}, {"RasteringFailed"}, {"Rastering"}),
+    "XRF":     ({"FScan"}, set(), set()),
+    "Collect": ({"Collected", "Characterized"},
+                {"CollectionFailed"}, {"Collecting", "Characterizing"}),
+    "Process": ({"Characterized"}, set(), set()),
+}
+# events that start a new run (a mount cycle)
+_RUN_START = {"Mounting", "Mounted", "AutomationStart"}
+# generic run-level failures (carry the error message)
+_RUN_FAILURES = {"Failed", "AutomationFailed", "AutomationCancelled"}
+# which stage having an OK event implies a collection type
+_TYPE_BY_STAGE = {"Collect": "Rotation", "Raster": "Raster", "XRF": "XRF"}
+
+
+class SampleRunStage(BaseModel):
+    stage: str
+    status: StageStatus = StageStatus.UNKNOWN
+    message: Optional[str] = None
+
+
+class SampleRunRecord(BaseModel):
+    label: str
+    type: str = "Run"
+    when: Optional[str] = None          # ISO timestamp (str for easy transport)
+    res: Optional[str] = None
+    stages: list[SampleRunStage]
+    error: Optional[str] = None
+
+
+class SampleHistory(BaseModel):
+    sample_id: int
+    sample_name: str
+    summary: str = ""
+    runs: list[SampleRunRecord] = []        # newest first
+    latest_stages: list[SampleRunStage] = []  # the per-row 6-LED summary
+
+
+def _ename(ev) -> str:
+    et = getattr(ev, "event_type", ev)
+    return getattr(et, "value", None) or getattr(et, "name", None) or str(et)
+
+
+def _when(ev) -> Optional[str]:
+    ts = getattr(ev, "timestamp", None)
+    if ts is None:
+        return None
+    return ts.isoformat() if hasattr(ts, "isoformat") else str(ts)
+
+
+def _stages_for_run(run_events: list) -> tuple[list[SampleRunStage], Optional[str], str]:
+    names = [_ename(e) for e in run_events]
+    run_error = None
+    run_type = "Mount"
+    stages: list[SampleRunStage] = []
+    for stage in STAGES:
+        ok, fail, prog = _STAGE_EVENTS[stage]
+        status, message = StageStatus.SKIP, None
+        for ev, nm in zip(run_events, names):
+            if nm in fail:
+                status, message = StageStatus.ERROR, getattr(ev, "comment", None)
+            elif nm in ok and status is not StageStatus.ERROR:
+                status = StageStatus.OK
+            elif nm in prog and status is StageStatus.SKIP:
+                status = StageStatus.RUNNING
+        if status is not StageStatus.SKIP and stage in _TYPE_BY_STAGE:
+            run_type = _TYPE_BY_STAGE[stage]
+        if status is StageStatus.ERROR and message:
+            run_error = f"{stage} failed: {message}"
+        stages.append(SampleRunStage(stage=stage, status=status, message=message))
+    # generic run-level failure (no stage-specific one captured)
+    if run_error is None:
+        for ev, nm in zip(run_events, names):
+            if nm in _RUN_FAILURES:
+                run_error = getattr(ev, "comment", None) or nm
+    return stages, run_error, run_type
+
+
+def build_sample_history(sample_id: int, sample_name: str, events: list) -> SampleHistory:
+    """Group sorted events into runs and derive per-stage outcomes."""
+    evs = sorted(events or [], key=lambda e: getattr(e, "timestamp", 0) or 0)
+    runs_events: list[list] = []
+    cur: Optional[list] = None
+    for ev in evs:
+        nm = _ename(ev)
+        start = nm == "Mounting" or nm == "AutomationStart" or (
+            nm == "Mounted" and cur is None)
+        if start:
+            if cur:
+                runs_events.append(cur)
+            cur = [ev]
+        else:
+            if cur is None:
+                cur = []
+            cur.append(ev)
+    if cur:
+        runs_events.append(cur)
+
+    records: list[SampleRunRecord] = []
+    n_err = 0
+    for i, rev in enumerate(runs_events, start=1):
+        stages, err, rtype = _stages_for_run(rev)
+        if err:
+            n_err += 1
+        records.append(SampleRunRecord(
+            label=f"Run {i}", type=rtype, when=_when(rev[0]),
+            stages=stages, error=err))
+    records.reverse()   # newest first
+
+    latest = records[0].stages if records else [
+        SampleRunStage(stage=s, status=StageStatus.UNKNOWN) for s in STAGES]
+    last_when = records[0].when if records else None
+    summary = (f"{len(records)} run{'s' if len(records) != 1 else ''}"
+               + (f" · last {last_when[:16].replace('T', ' ')}" if last_when else "")
+               + (f" · {n_err} error{'s' if n_err != 1 else ''}" if n_err else ""))
+    return SampleHistory(sample_id=sample_id, sample_name=sample_name,
+                         summary=summary, runs=records, latest_stages=latest)