"""
Battle Trace: an experimental temporal replay of a session as a war between the
Agent and the Environment.
This is the LITERAL, time-based telling of the session — complementary to the
shell's byobu battle layer, which is the frozen, artistic telling. The shell
remembers the campaign as a folding screen; Battle Trace replays it as it
happened.
It reuses the SAME structured extraction the rest of the app already produces
(approaches, dead_ends, breakthroughs, gotchas, sentiment), so there is no
second parser and no new model call. Events map to combat:
approach tried -> the Agent makes a move (cmd)
dead end -> the Environment strikes (err): Agent resolve drops
gotcha -> a clash of blades (clash)
breakthrough -> a blow lands true (win): Environment resistance drops
final breakthrough -> the battle resolves (done)
The renderer is self-contained Canvas 2D (fine for session-sized traces),
sandboxed inside an iframe so it can never affect the main pipeline.
"""
from __future__ import annotations
import html
import json
def extraction_to_events(extraction: dict) -> list[dict]:
"""Convert a session extraction into an ordered Battle Trace event stream.
Events are ordered by their position along the session (0..1) where we have
it (dead ends, breakthroughs), and interleaved sensibly otherwise.
"""
events: list[dict] = []
approaches = extraction.get("approaches_tried", []) or []
dead_ends = extraction.get("dead_ends", []) or []
breakthroughs = extraction.get("breakthroughs", []) or []
gotchas = extraction.get("gotchas", []) or []
# approaches kick off near the start, spaced across the first half
for i, a in enumerate(approaches):
if not isinstance(a, dict):
continue
pos = 0.05 + (i / max(1, len(approaches))) * 0.4
label = str(a.get("approach", "a move"))[:72]
events.append({"pos": pos, "side": "A", "kind": "cmd", "label": label})
# dead ends at their real positions — the Environment strikes
for d in dead_ends:
if not isinstance(d, dict):
continue
pos = float(d.get("position", 0.5))
label = str(d.get("what_happened", "a wall"))[:72]
events.append({"pos": pos, "side": "E", "kind": "err", "label": label})
# gotchas as clashes, spread through the middle
for i, g in enumerate(gotchas):
pos = 0.3 + (i / max(1, len(gotchas))) * 0.5
label = str(g)[:72]
events.append({"pos": pos, "side": "A", "kind": "clash", "label": label})
# breakthroughs at their positions — blows land
for i, b in enumerate(breakthroughs):
if not isinstance(b, dict):
continue
pos = float(b.get("position", 0.85))
label = str(b.get("what_worked", "a breakthrough"))[:72]
kind = "done" if i == len(breakthroughs) - 1 else "win"
events.append({"pos": pos, "side": "A", "kind": kind, "label": label})
events.sort(key=lambda e: e["pos"])
# assign a monotonic time axis from positions
for i, e in enumerate(events):
e["t"] = round(1.0 + e["pos"] * 12.0, 2)
if not events:
events = [{"pos": 0.5, "side": "A", "kind": "info", "label": "a quiet session", "t": 6.0}]
# ensure the last event resolves the battle
if events[-1]["kind"] not in ("done", "err"):
events[-1]["kind"] = "done"
return events
# The renderer: a trimmed, self-contained Canvas 2D battle. No external deps.
# Events are injected as JSON; everything else is static.
_BATTLE_HTML = r"""
⬡ THE AGENT (you) resolve 100
THE ENVIRONMENT ⬡ resistance 100
t+0.0sthe slug replays the battle, from the first move...
"""
def render_battle_trace(extraction: dict, height: int = 420) -> str:
"""Return an iframe hosting the Battle Trace replay for this extraction.
Sandboxed iframe so it cannot affect the main app. Best-effort: any failure
returns an empty string and the accordion simply shows nothing.
"""
try:
events = extraction_to_events(extraction)
doc = _BATTLE_HTML.replace("__EVENTS_JSON__", json.dumps(events))
escaped = html.escape(doc, quote=True)
return (
f''
)
except Exception:
return ""