#!/usr/bin/env python3 """ Bannerlord Session Trace Logger — First-Replayable Training Material Captures one Bannerlord session as a replayable trace: - Timestamps on every cycle - Actions executed with success/failure - World-state evidence (screenshots, Steam stats) - Hermes session/log ID mapping Storage: ~/.timmy/traces/bannerlord/trace_.jsonl Manifest: ~/.timmy/traces/bannerlord/manifest_.json Each JSONL line is one ODA cycle with full context. The manifest bundles metadata for replay/eval. """ from __future__ import annotations import json import time import uuid from dataclasses import dataclass, field, asdict from datetime import datetime, timezone from pathlib import Path from typing import Optional # Storage root — local-first under ~/.timmy/ DEFAULT_TRACE_DIR = Path.home() / ".timmy" / "traces" / "bannerlord" @dataclass class CycleTrace: """One ODA cycle captured in full.""" cycle_index: int timestamp_start: str timestamp_end: str = "" duration_ms: int = 0 # Observe screenshot_path: str = "" window_found: bool = False screen_size: list[int] = field(default_factory=lambda: [1920, 1080]) mouse_position: list[int] = field(default_factory=lambda: [0, 0]) playtime_hours: float = 0.0 players_online: int = 0 is_running: bool = False # Decide actions_planned: list[dict] = field(default_factory=list) decision_note: str = "" # Act actions_executed: list[dict] = field(default_factory=list) actions_succeeded: int = 0 actions_failed: int = 0 # Metadata hermes_session_id: str = "" hermes_log_id: str = "" harness_session_id: str = "" def to_dict(self) -> dict: return asdict(self) @dataclass class SessionManifest: """Top-level metadata for a captured session trace.""" trace_id: str harness_session_id: str hermes_session_id: str hermes_log_id: str game: str = "Mount & Blade II: Bannerlord" app_id: int = 261550 started_at: str = "" finished_at: str = "" total_cycles: int = 0 total_actions: int = 0 total_succeeded: int = 0 total_failed: int = 0 trace_file: str = "" trace_dir: str = "" replay_command: str = "" eval_note: str = "" def to_dict(self) -> dict: return asdict(self) class BannerlordTraceLogger: """ Captures a single Bannerlord session as a replayable trace. Usage: logger = BannerlordTraceLogger(hermes_session_id="abc123") logger.start_session() cycle = logger.begin_cycle(0) # ... populate cycle fields ... logger.finish_cycle(cycle) manifest = logger.finish_session() """ def __init__( self, trace_dir: Optional[Path] = None, harness_session_id: str = "", hermes_session_id: str = "", hermes_log_id: str = "", ): self.trace_dir = trace_dir or DEFAULT_TRACE_DIR self.trace_dir.mkdir(parents=True, exist_ok=True) self.trace_id = f"bl_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:6]}" self.harness_session_id = harness_session_id or str(uuid.uuid4())[:8] self.hermes_session_id = hermes_session_id self.hermes_log_id = hermes_log_id self.trace_file = self.trace_dir / f"trace_{self.trace_id}.jsonl" self.manifest_file = self.trace_dir / f"manifest_{self.trace_id}.json" self.cycles: list[CycleTrace] = [] self.started_at: str = "" self.finished_at: str = "" def start_session(self) -> str: """Begin a trace session. Returns trace_id.""" self.started_at = datetime.now(timezone.utc).isoformat() return self.trace_id def begin_cycle(self, cycle_index: int) -> CycleTrace: """Start recording one ODA cycle.""" cycle = CycleTrace( cycle_index=cycle_index, timestamp_start=datetime.now(timezone.utc).isoformat(), harness_session_id=self.harness_session_id, hermes_session_id=self.hermes_session_id, hermes_log_id=self.hermes_log_id, ) return cycle def finish_cycle(self, cycle: CycleTrace) -> None: """Finalize and persist one cycle to the trace file.""" cycle.timestamp_end = datetime.now(timezone.utc).isoformat() # Compute duration try: t0 = datetime.fromisoformat(cycle.timestamp_start) t1 = datetime.fromisoformat(cycle.timestamp_end) cycle.duration_ms = int((t1 - t0).total_seconds() * 1000) except (ValueError, TypeError): cycle.duration_ms = 0 # Count successes/failures cycle.actions_succeeded = sum( 1 for a in cycle.actions_executed if a.get("success", False) ) cycle.actions_failed = sum( 1 for a in cycle.actions_executed if not a.get("success", True) ) self.cycles.append(cycle) # Append to JSONL with open(self.trace_file, "a") as f: f.write(json.dumps(cycle.to_dict()) + "\n") def finish_session(self) -> SessionManifest: """Finalize the session and write the manifest.""" self.finished_at = datetime.now(timezone.utc).isoformat() total_actions = sum(len(c.actions_executed) for c in self.cycles) total_succeeded = sum(c.actions_succeeded for c in self.cycles) total_failed = sum(c.actions_failed for c in self.cycles) manifest = SessionManifest( trace_id=self.trace_id, harness_session_id=self.harness_session_id, hermes_session_id=self.hermes_session_id, hermes_log_id=self.hermes_log_id, started_at=self.started_at, finished_at=self.finished_at, total_cycles=len(self.cycles), total_actions=total_actions, total_succeeded=total_succeeded, total_failed=total_failed, trace_file=str(self.trace_file), trace_dir=str(self.trace_dir), replay_command=( f"python -m nexus.bannerlord_harness --mock --replay {self.trace_file}" ), eval_note=( "To replay: load this trace, re-execute each cycle's actions_planned " "against a fresh harness in mock mode, compare actions_executed outcomes. " "Success metric: >=90% action parity between original and replay runs." ), ) with open(self.manifest_file, "w") as f: json.dump(manifest.to_dict(), f, indent=2) return manifest @classmethod def load_trace(cls, trace_file: Path) -> list[dict]: """Load a trace JSONL file for replay or analysis.""" cycles = [] with open(trace_file) as f: for line in f: line = line.strip() if line: cycles.append(json.loads(line)) return cycles @classmethod def load_manifest(cls, manifest_file: Path) -> dict: """Load a session manifest.""" with open(manifest_file) as f: return json.load(f) @classmethod def list_traces(cls, trace_dir: Optional[Path] = None) -> list[dict]: """List all available trace sessions.""" d = trace_dir or DEFAULT_TRACE_DIR if not d.exists(): return [] traces = [] for mf in sorted(d.glob("manifest_*.json")): try: manifest = cls.load_manifest(mf) traces.append(manifest) except (json.JSONDecodeError, IOError): continue return traces