Analyze gameplay and transfer mouse sensitivity between games

Game Sensitivity Coach

An agent skill that discovers your mouse settings, carries a familiar sensitivity into another game, and reviews ordinary gameplay to recommend adjustments. You play; your agent handles recording, image inspection, evidence, and calculations.

Built from practical work with THE FINALS, Battlefield 6, and Holdfast. The workflow is game-independent; capture support, file formats, and each game's sensitivity math still need verification.

An example from THE FINALS. The agent reviews the frames and input logs itself. This optional viewer lets you inspect its evidence; you do not need to label shots or fill in forms.

What it does

When you authorize a change, the agent prepares an exact patch, backs up the original, applies it with the game closed, and verifies the result. It can then compare another sample and keep or roll back the change within your agreed limits. There is no universal “best” number: retaining a well-supported baseline is a valid verdict.

Install

Download the latest skill ZIP, then extract it. Keep the entire game-sensitivity-coach folder together, including scripts, references, and assets. Each release also includes a SHA-256 checksum.

Alternatively, clone directly into your agent's skill directory. For a personal Codex installation:

git clone https://github.com/DenisSergeevitch/game-sensitivity-coach.git ~/.agents/skills/game-sensitivity-coach

On Windows, ~ means your user home directory. Follow the current Codex skill documentation or Claude Code skill documentation if your host uses a different setup. The package follows the Agent Skills format; agents/openai.yaml supplies optional Codex display metadata.

You need:

• An agent that can run local commands, read files, and inspect images.
• Python 3.8 or newer. The bundled scripts use the Python standard library; no Python packages or separate AI API key are required.
• FFmpeg for recording video and exporting evidence. Put it on PATH, or pass its location with --ffmpeg. Use a build with libx264, H.264 decoding, and PNG output; see the FFmpeg download page.
• Windows on the gaming computer for bundled live capture. Offline reports, conversion, and file helpers are portable Python. A cloud-only agent needs existing session files or a connection to an authorized local runner; it cannot see your PC's monitor by itself. For WSL, launch the recorder with native Windows Python.

Start with a message

In Codex, mention $game-sensitivity-coach. In Claude Code, invoke /game-sensitivity-coach, then describe the task. For example:

Find my current THE FINALS mouse settings. I will play normally for eight minutes; record that session, inspect my sniper aim, and tell me which sensitivity to keep or test. Handle the analysis yourself.

Transfer my current THE FINALS sensitivity into Battlefield 6. Find both profiles and explain the exact target value and assumptions. Prepare the patch for me to review.

Apply the proposed scoped setting after I close the game. Back up the file, then compare another five-minute session. Ask before making any further change.

The agent discovers what it can instead of presenting a setup form. It may need one essential clarification if neither files nor images establish a required fact. A request for a recommendation alone does not authorize a settings change; an explicit application request does.

How the observation works

At launch, the recorder chooses the monitor most covered by a visible matching game window, unless the agent supplies --monitor. It records that selected monitor while the matching game is foreground and receives Windows mouse Raw Input. It saves video, frame acquisition times, mouse counts/buttons, metadata, and an integrity manifest. Capture pauses when that game loses focus; monitor selection does not follow later window moves. Visible overlays and notifications can appear in a full-monitor recording.

The report aligns click candidates with actual encoded frames. Your agent exports those images, inspects target-to-crosshair motion and weapon state, and writes an assessment with frame references. It also samples combat beyond clicks, since holding fire can produce multiple shots. Pixel estimates and raw mouse counts remain distinct.

The scripts organize evidence; your active host agent supplies the visual reasoning. They do not launch an unattended AI service or automatically detect every enemy. Without an image-capable agent, the report remains descriptive and has no invented sensitivity verdict.

A recommendation records its baseline, proposed value, relevant weapon/optic, supporting observations, exclusions, and limitations. New settings are tested one at a time under comparable conditions. The original recordings and previous settings remain traceable throughout the loop.

Included tools

These commands are primarily for the agent. Run them from this folder, or use absolute script paths. Use python3 or py -3 if that is your Python command.

python scripts/discover_settings.py --game "THE FINALS"
python scripts/capture_gameplay.py --game "THE FINALS" --seconds 480 --wait-for-game 60 --output-root "../GameSensitivity/sessions" --review-on-stop
python scripts/review_session.py "SESSION" --verify-manifest
python scripts/export_evidence.py "SESSION" --overview-count 12

The agent can pass an automatically discovered profile with --profile. Unknown settings stay unknown. The recorder prints the actual session path; SESSION above is a placeholder for that path. --review-on-stop generates descriptive evidence; the agent still needs to inspect it and author the recommendation.

By default, the exporter selects up to 12 overview frames and six click candidates spread across recorded-frame/candidate positions, with six actual frames before and after each selected click where available. Overview images do not screen every candidate or every combat interval. To screen all candidates, the agent reads their count/IDs from analysis.json and supplies that count with --click-count or selects every ID with --click. It then expands useful sequences with --before-frames and --after-frames. Each export requires a fresh --output directory; the default is SESSION/evidence. For example, after selecting a valid candidate ID:

python scripts/export_evidence.py "SESSION" --click click-00003 --before-frames 12 --after-frames 6 --output "SESSION/evidence-click-00003"

After visual analysis:

python scripts/review_session.py "SESSION" --assessment "ASSESSMENT.json" --verify-manifest

To change the duration of a running recording, use its existing control file:

python scripts/capture_gameplay.py control "SESSION/control.json" --extend-seconds 120
python scripts/capture_gameplay.py control "SESSION/control.json" --stop

--extend-seconds adds to the existing deadline; --total-seconds sets the total duration from the original start. Send changes before that deadline expires and check for control_applied or the updated capture-status.json. Extensions honor the fixed --max-total-seconds cap set at launch: 3,600 seconds by default, at most 14,400. Initial --seconds must be at least one second and no greater than that cap. The run timer includes initialization and focus pauses. Optional --wait-for-game adds a separate foreground wait of up to 3,600 seconds before the session starts. The recorder never silently restarts to extend a run.

The conversion example is synthetic test data, not a preset for a real game:

python scripts/convert_sensitivity.py examples/conversion.synthetic.json

It yields target sensitivity 5 from source 10 under the example's explicit calibration. Real transfers need documented or measured angular scaling; identical slider percentages do not establish equivalence. Unknown DPI can support a stated same-DPI assumption, but cannot establish absolute cm/360.

For a supported active configuration, the application flow is:

python scripts/settings_file.py inspect "ACTIVE.ini" --section Controls --key Sensitivity
python scripts/settings_file.py plan "ACTIVE.ini" --section Controls --change Sensitivity 3.00 5.00 --output "PLAN.json"
python scripts/settings_file.py apply "PLAN.json" --reviewed --game-closed
python scripts/settings_file.py rollback "RECEIPT.json" --game-closed

Those values and filenames are illustrative. The agent generates the real plan from discovered settings. It supplies the check flags only after verifying authorization and that the game/config writer is closed. See application and recovery for receipts, interruption handling, and guarded rollback.

Supported scope

The recorder does not inspect game memory, inject input, hook rendering, or aim for you. Settings inspection reads on-disk configuration separately. Capture availability and game rules still vary; the skill does not bypass capture restrictions or claim approval from any game's anti-cheat system.

Session artifacts stay in your chosen output directory. Images consumed by the host agent follow that host's model and privacy settings. Captures, real saves, account profiles, and private backup files are excluded from the release package.

Repository guide

• SKILL.md: agent entry point and autonomous workflow.
• Discovery, observation, conversion, and application: detailed procedures.
• Game examples: lessons from THE FINALS → Battlefield 6, Holdfast, and a sniper review that retained 72% scoped sensitivity. Historical values are examples, not recommendations for a new gamer.
• Data contracts and examples: profiles, visual observations, assessments, conversion requests, and iteration journals.
• scripts/: capture, reports, evidence export, discovery, conversion, configuration changes, and release packaging.
• tests/: synthetic fixtures and regression tests; running the suite does not record your desktop or change game settings.

Test and package

python -m unittest discover -s tests -v
python scripts/package_release.py --output ../dist/game-sensitivity-coach.zip

The package builder includes only distributable source, documentation, examples, tests, and the supplied screenshot. It writes a reproducible ZIP with one top-level skill folder, an internal file/hash manifest, and an adjacent SHA-256 file. Runtime outputs and caches are excluded. Use --replace to replace a previous release deliberately.

The synthetic encoder test runs when FFmpeg is available and otherwise reports a skip. Real gameplay quality, game-version compatibility, and any proposed performance improvement require a fresh observation on the target computer.

scripts/discover_settings.py

"""List likely settings files without opening game memory or editing files."""
import argparse
import json
import os
import re
import stat
from pathlib import Path

HINTS = {
    'thefinals': [('local', 'Discovery/Saved/SaveGames/EmbarkOptionSaveGame.sav'),
                  ('local', 'Discovery/Saved/Config/WindowsClient/GameUserSettings.ini')],
    'battlefield6': [('documents', 'Battlefield 6/settings')],
    'holdfastnationsatwar': [('low', 'Anvil Game Studio/Holdfast NaW/HoldfastOptions.ini')],
    'holdfast': [('low', 'Anvil Game Studio/Holdfast NaW/HoldfastOptions.ini')],
}


def normalized(text):
    return re.sub(r'[^a-z0-9]', '', text.lower())


def known_roots():
    home = Path.home()
    roots = {'local': Path(os.environ.get('LOCALAPPDATA', home/'AppData/Local')),
             'roaming': Path(os.environ.get('APPDATA', home/'AppData/Roaming')),
             'low': home/'AppData/LocalLow', 'documents': home/'Documents',
             'saved': home/'Saved Games', 'config': home/'.config'}
    if os.name == 'nt':
        # Resolve redirected Documents (e.g. OneDrive) through the Windows shell.
        import ctypes
        buffer = ctypes.create_unicode_buffer(32768)
        if ctypes.windll.shell32.SHGetFolderPathW(None, 5, None, 0, buffer) == 0:
            roots['documents'] = Path(buffer.value)
    return roots


def reparse(path):
    try:
        s = path.lstat()
        return stat.S_ISLNK(s.st_mode) or bool(getattr(s, 'st_file_attributes', 0) & 0x400)
    except OSError:
        return True


def discover(game, explicit_roots=None, aliases=None, max_files=5000):
    roots = known_roots()
    explicit_roots = [Path(p).expanduser().resolve() for p in (explicit_roots or [])]
    needles = {normalized(x) for x in [game] + list(aliases or []) if normalized(x)}
    if not needles:
        raise ValueError('A nonempty game name or alias is required')
    candidates = list(explicit_roots)
    errors = []
    if not explicit_roots:
        for root_name, relative in HINTS.get(normalized(game), []):
            candidates.append(roots[root_name]/relative)
        for root in roots.values():
            if not root.is_dir(): continue
            try:
                candidates.extend(p for p in root.iterdir() if p.is_dir() and
                                  any(n in normalized(p.name) for n in needles))
            except OSError as error:
                errors.append({'path':str(root), 'error':str(error)})
    results, seen = [], set()
    examined = 0
    truncated = False
    def eligible(path):
        return path.suffix.lower() in ('.ini','.cfg','.conf','.json','.xml','.sav') or path.name.lower().startswith('profsave')
    def add(path):
        key = str(path.resolve())
        if key in seen or not eligible(path): return
        seen.add(key)
        try:
            s=path.stat()
            results.append({'path':key,'bytes':s.st_size,'modified_unix_s':s.st_mtime,
                            'hint':'Candidate only; verify active profile, keys and encoding before use.'})
        except OSError as error:
            errors.append({'path':str(path),'error':str(error)})
    for root in candidates:
        if truncated: break
        if not root.exists(): continue
        if root.is_file(): add(root); continue
        if reparse(root):
            errors.append({'path':str(root),'error':'Reparse/symlink root skipped; supply the resolved local directory.'})
            continue
        def onerror(error): errors.append({'path':getattr(error,'filename',''), 'error':str(error)})
        for directory, dirs, files in os.walk(str(root), followlinks=False, onerror=onerror):
            depth=len(Path(directory).relative_to(root).parts)
            dirs[:] = [d for d in dirs if depth < 6 and not reparse(Path(directory)/d)
                       and d.lower() not in ('node_modules','.git','cache','shadercache')]
            # Bound directories as well as files to keep large install scans finite.
            examined += 1
            for name in files:
                examined += 1
                if examined > max_files:
                    truncated=True; break
                path=Path(directory)/name
                if not reparse(path): add(path)
            if examined > max_files:
                truncated=True; break
    return {'schema':'game-sensitivity/discovery-v1','game':game,'candidates':sorted(results,key=lambda x:x['path']),
            'examined_entries':examined,'truncated':truncated,'errors':errors,
            'note':'Read-only location discovery. Paths are hints, not proof of active settings. Unsupported games may need an install root, publisher alias, launcher profile, registry or in-game settings view.'}


def main():
    parser=argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--game',required=True)
    parser.add_argument('--root',action='append',help='Search this bounded directory instead of default user-folder hints')
    parser.add_argument('--alias',action='append',default=[])
    parser.add_argument('--max-files',type=int,default=5000)
    args=parser.parse_args()
    if not 1 <= args.max_files <= 50000:parser.error('max-files must be between 1 and 50000')
    try: print(json.dumps(discover(args.game,args.root,args.alias,args.max_files),indent=2))
    except (OSError,ValueError) as error:parser.exit(2,str(error)+'\n')


if __name__=='__main__':main()

scripts/capture_gameplay.py

#!/usr/bin/env python3
"""Bounded Windows gameplay recording. Imports no capture API for --help/control.

Examples:
  python capture_gameplay.py --game "Example Game" --seconds 480 --wait-for-game 60
  python capture_gameplay.py control SESSION/control.json --extend-seconds 480
  python capture_gameplay.py control SESSION/control.json --stop
"""
import argparse
import datetime as dt
import json
import math
import os
from pathlib import Path
import sys
import time
import unicodedata
import uuid

from aim_observer.profiles import KNOWN_SETTINGS, load_profile, validate_settings


CONTROL_SCHEMA = "game-sensitivity/capture-control-v1"
HARD_LIMIT_SECONDS = 14400


def atomic_json(path, value):
    path = Path(path)
    temporary = path.with_name(path.name + "." + uuid.uuid4().hex + ".tmp")
    temporary.write_text(json.dumps(value, indent=2, allow_nan=False) + "\n", encoding="utf-8")
    temporary.replace(path)


def bounded_number(value):
    value = float(value)
    if not math.isfinite(value) or not 1 <= value <= HARD_LIMIT_SECONDS:
        raise argparse.ArgumentTypeError("Duration must be between 1 and 14400 seconds")
    return value


class CaptureDeadline:
    """A monotonic deadline with replay-safe, explicitly bounded extensions."""
    def __init__(self, start, seconds, max_total_seconds):
        if not (1 <= seconds <= max_total_seconds <= HARD_LIMIT_SECONDS):
            raise ValueError("Require 1 <= seconds <= max-total-seconds <= 14400")
        self.start = start
        self.deadline = start + seconds
        self.max_total_seconds = max_total_seconds
        self.stopped = False
        self.seen = set()

    def apply(self, document):
        if not isinstance(document, dict) or document.get("schema") != CONTROL_SCHEMA:
            raise ValueError("Unsupported capture control document")
        request_id = document.get("request_id")
        if not isinstance(request_id, str) or not request_id.strip():
            raise ValueError("Control request_id must be nonempty text")
        if request_id in self.seen:
            return None
        action = document.get("action")
        if action == "stop":
            self.stopped = True
        elif action in ("extend", "set_total"):
            seconds = document.get("seconds")
            if isinstance(seconds, bool) or not isinstance(seconds, (int, float)) or not math.isfinite(seconds) or seconds < 1:
                raise ValueError("Control seconds must be finite and at least 1")
            deadline = self.deadline + seconds if action == "extend" else self.start + seconds
            if deadline > self.start + self.max_total_seconds:
                raise ValueError("Control request exceeds the recording's fixed max-total-seconds")
            self.deadline = deadline
        else:
            raise ValueError("Control action must be stop, extend, or set_total")
        self.seen.add(request_id)
        return action


def build_parser():
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--game", help="Game name; may instead come from --profile")
    parser.add_argument("--window-title", help="Matching visible window title substring; defaults to game name")
    parser.add_argument("--seconds", required=True, type=bounded_number, help="Initial wall-clock recording duration (1-14400 seconds)")
    parser.add_argument("--max-total-seconds", type=bounded_number, default=3600,
                        help="Fixed total-duration cap for control-file extensions (default 3600)")
    parser.add_argument("--wait-for-game", type=float, default=0,
                        help="Wait up to this many seconds for the matching game to become foreground (0-3600)")
    parser.add_argument("--output-root", type=Path, default=Path.cwd() / "sessions")
    parser.add_argument("--control-file", type=Path, help="Control JSON path; default SESSION/control.json")
    parser.add_argument("--profile", type=Path, help="Agent-discovered game-sensitivity/profile-v1 JSON")
    parser.add_argument("--dpi", type=int, default=None, help="Known DPI; otherwise preserve profile value or null")
    parser.add_argument("--look-sensitivity", type=float, default=None, help="Known game value; otherwise preserve profile value or null")
    parser.add_argument("--scoped-multiplier", type=float, default=None, help="Known multiplier; otherwise preserve profile value or null")
    parser.add_argument("--normal-zoom-multiplier", type=float, default=None, help="Known multiplier; otherwise preserve profile value or null")
    parser.add_argument("--monitor", type=int, help="Explicit zero-based monitor index; default detects matching game window")
    parser.add_argument("--fps", type=int, choices=(15, 30, 60), default=30)
    parser.add_argument("--max-width", type=int, choices=(1280, 1920, 2560, 3840), default=1920)
    parser.add_argument("--ffmpeg", help="FFmpeg executable; default searches PATH")
    parser.add_argument("--hardware", action="store_true", help="Use NVIDIA h264_nvenc instead of CPU libx264")
    parser.add_argument("--review-on-stop", action="store_true", help="Build the offline descriptive report after finalizing")
    return parser


def resolve_context(args):
    profile = load_profile(args.profile)
    game = args.game or profile.get("game")
    if not isinstance(game, str) or not game.strip():
        raise ValueError("Supply --game or a profile containing game")
    title = args.window_title if args.window_title is not None else profile.get("window_title", game)
    if (not isinstance(title, str)
            or not "".join(char for char in title if unicodedata.category(char) != "Cf").strip()):
        raise ValueError("A nonempty game window title is required")
    settings = {field: None for field in KNOWN_SETTINGS}
    settings.update(profile.get("settings", {}))
    sources = dict(profile.get("setting_sources", {}))
    for field in KNOWN_SETTINGS:
        explicit = getattr(args, field)
        if explicit is not None:
            settings[field] = explicit
            sources[field] = {"status": "provided", "source": "command-line",
                              "observed_at_utc": dt.datetime.now(dt.timezone.utc).isoformat()}
        elif field not in sources:
            sources[field] = {"status": "unknown" if settings[field] is None else "provided",
                              "source": None if settings[field] is None else "profile"}
    validate_settings(settings)
    context = dict(profile.get("context", {}))
    if args.profile:
        context.update(profile_schema=profile.get("schema"), profile_evidence=profile.get("evidence", []))
    return dict(game=game.strip(), title=title.strip(), settings=settings,
                setting_sources=sources, context=context,
                settings_source=profile.get("settings_source", profile.get("source")))


def wait_for_game(gate, seconds, clock=time.monotonic, sleep=time.sleep):
    """Wait only within a fixed bound; never capture while waiting."""
    deadline = clock() + seconds
    while True:
        if gate.active_handle():
            return
        remaining = deadline - clock()
        if remaining <= 0:
            raise RuntimeError("The matching game did not become foreground within --wait-for-game")
        sleep(min(.1, remaining))


def run_capture(args):
    context = resolve_context(args)
    if args.seconds > args.max_total_seconds:
        raise ValueError("--seconds cannot exceed --max-total-seconds")
    if not math.isfinite(args.wait_for_game) or not 0 <= args.wait_for_game <= 3600:
        raise ValueError("--wait-for-game must be between 0 and 3600 seconds")
    if os.name != "nt":
        raise RuntimeError("Live gameplay capture requires Windows. Offline review and evidence export work on other platforms.")
    from aim_observer.capture import (ForegroundGate, enable_dpi_awareness, find_ffmpeg,
                                      find_game_monitor, monitors)
    from aim_observer.recorder import RecordingSession
    ffmpeg = find_ffmpeg(args.ffmpeg)
    enable_dpi_awareness()
    if args.wait_for_game:
        print(json.dumps({"status": "waiting_for_game", "game": context["game"],
                          "window_title": context["title"], "timeout_s": args.wait_for_game}), flush=True)
        wait_for_game(ForegroundGate(context["title"]), args.wait_for_game)
    displays = monitors()
    if not displays:
        raise RuntimeError("No display is available")
    selected = args.monitor if args.monitor is not None else find_game_monitor(displays, context["title"])
    if selected is None:
        raise RuntimeError("No visible matching game window was found. Open the game, use --wait-for-game, or supply --monitor.")
    if not 0 <= selected < len(displays):
        raise ValueError("--monitor is outside the available monitor indices")
    session = RecordingSession(args.output_root, displays[selected], fps=args.fps,
                               max_width=args.max_width, ffmpeg=ffmpeg,
                               hardware=args.hardware, **context)
    control = args.control_file or session.folder / "control.json"
    if control.exists():
        raise ValueError("Control file already exists; choose a fresh --control-file")
    control.parent.mkdir(parents=True, exist_ok=True) if args.control_file else None
    # The cap is fixed at launch. Extensions change the deadline in this same
    # process and session; they never restart capture or invent gap-filling frames.
    started = time.monotonic()
    deadline = CaptureDeadline(started, args.seconds, args.max_total_seconds)
    last_content, next_status, control_error = None, 0, None
    stop_reason = "deadline"
    try:
        folder = session.start()
        # Initialization belongs to the bounded run but may take some seconds.
        initial = {"schema": CONTROL_SCHEMA, "request_id": uuid.uuid4().hex,
                   "action": "set_total", "seconds": args.seconds}
        atomic_json(control, initial)
        session.metadata["runtime"] = {
            "initial_seconds": args.seconds, "max_total_seconds": args.max_total_seconds,
            "duration_clock": "monotonic elapsed time including initialization and focus pauses",
            "control_file": str(control.resolve()), "control_history": [],
        }
        print(json.dumps({"status": "recording", "session": str(folder.resolve()),
                          "control_file": str(control.resolve()), "max_total_seconds": args.max_total_seconds}), flush=True)
        while not deadline.stopped and time.monotonic() < deadline.deadline:
            now = time.monotonic()
            changed = False
            try:
                content = control.read_text(encoding="utf-8-sig")
                if content != last_content:
                    changed = True
                    last_content = content
                    document = json.loads(content)
                    action = deadline.apply(document)
                    control_error = None
                    if action:
                        session.metadata["runtime"]["control_history"].append(document)
                        print(json.dumps({"status": "control_applied", "request_id": document["request_id"],
                                          "action": action, "planned_total_seconds": deadline.deadline-deadline.start}), flush=True)
            except (OSError, ValueError, TypeError, OverflowError) as error:
                control_error = str(error)
                if changed or now >= next_status:
                    print(json.dumps({"status": "control_warning", "error": str(error)}), flush=True)
            info = session.status()
            if session.stop_event.is_set() or info["errors"]:
                stop_reason = "recording_error"
                break
            if now >= next_status:
                snapshot = dict(info, status="recording", session=str(folder.resolve()),
                                control_file=str(control.resolve()),
                                remaining_s=max(0, deadline.deadline-now),
                                planned_total_seconds=deadline.deadline-deadline.start,
                                max_total_seconds=deadline.max_total_seconds,
                                control_error=control_error)
                atomic_json(folder / "capture-status.json", snapshot)
                next_status = now + 1
            session.stop_event.wait(min(.1, max(0, deadline.deadline-time.monotonic())))
        if deadline.stopped:
            stop_reason = "control_stop"
    except KeyboardInterrupt:
        stop_reason = "keyboard_interrupt"
    except Exception as error:
        stop_reason = "startup_or_runtime_error"
        session._fail("Capture runtime: " + str(error))
        raise
    finally:
        if session.folder.exists():
            session.metadata.setdefault("runtime", {})["stop_reason"] = stop_reason
            session.stop()
            atomic_json(session.folder / "capture-status.json", {
                "status": "stopped", "stop_reason": stop_reason,
                "session": str(session.folder.resolve()), "errors": list(session.errors),
                "files_finalized": session.metadata.get("files_finalized", False),
                "frames": session.frame_count,
            })
    if args.review_on_stop and session.metadata.get("files_finalized"):
        from aim_observer.review import generate_review
        generate_review(session.folder)
    print(json.dumps({"status": "stopped", "session": str(session.folder.resolve()),
                      "frames": session.frame_count, "input_events": session.input_count,
                      "errors": session.errors, "stop_reason": stop_reason}), flush=True)
    return 0 if session.frame_count and not session.errors and session.metadata.get("files_finalized") else 2


def send_control(argv):
    parser = argparse.ArgumentParser(description="Update one existing recording's bounded local control file")
    parser.add_argument("control_file", type=Path)
    action = parser.add_mutually_exclusive_group(required=True)
    action.add_argument("--stop", action="store_true")
    action.add_argument("--extend-seconds", type=bounded_number)
    action.add_argument("--total-seconds", type=bounded_number)
    args = parser.parse_args(argv)
    if not args.control_file.is_file():
        raise ValueError("The recording control file does not exist")
    existing = json.loads(args.control_file.read_text(encoding="utf-8-sig"))
    if not isinstance(existing, dict) or existing.get("schema") != CONTROL_SCHEMA:
        raise ValueError("The target is not a gameplay recording control file")
    document = {"schema": CONTROL_SCHEMA, "request_id": uuid.uuid4().hex,
                "action": "stop" if args.stop else ("extend" if args.extend_seconds else "set_total")}
    if not args.stop:
        document["seconds"] = args.extend_seconds or args.total_seconds
    atomic_json(args.control_file, document)
    print(json.dumps({"status": "control_requested", "control_file": str(args.control_file.resolve()),
                      "request": document}), flush=True)
    return 0


def main(argv=None):
    argv = sys.argv[1:] if argv is None else argv
    try:
        return send_control(argv[1:]) if argv and argv[0] == "control" else run_capture(build_parser().parse_args(argv))
    except (OSError, ValueError, RuntimeError, OverflowError) as error:
        print("capture_gameplay: " + str(error), file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())

scripts/review_session.py

"""Build an offline report from a recorded session, optionally attaching a reviewed assessment.

Python 3.8+ and the standard library are sufficient. This command works from
existing recordings and never imports screen-capture or live mouse-input modules.
"""

import argparse
import hashlib
import json
import os
from pathlib import Path
import re
import sys
import tempfile

from aim_observer.review import analyze_session, generate_review, validate_coach_assessment


def verify_manifest(session):
    """Verify the recorder's original sizes/hashes without rewriting its manifest."""
    session = Path(session)
    with (session / "manifest.json").open("r", encoding="utf-8-sig") as source:
        manifest = json.load(source)
    required = {"session.json", "mouse.csv", "frames.csv"}
    allowed = required | {"video.mp4"}
    if not isinstance(manifest, dict) or not required.issubset(manifest):
        raise ValueError("manifest.json must include session.json, mouse.csv and frames.csv")
    if set(manifest) - allowed:
        raise ValueError("manifest.json contains unsupported paths; only original session files are allowed")
    if (session / "video.mp4").exists() and "video.mp4" not in manifest:
        raise ValueError("manifest.json omits the existing video.mp4")
    verified = []
    for name, expected in sorted(manifest.items()):
        if not isinstance(expected, dict):
            raise ValueError("manifest.json entry {} must be an object".format(name))
        size = expected.get("bytes")
        if isinstance(size, bool) or not isinstance(size, int) or size < 0:
            raise ValueError("manifest.json entry {} must contain a nonnegative byte count".format(name))
        digest = expected.get("sha256")
        if name in required or digest is not None:
            if not isinstance(digest, str) or not re.fullmatch(r"[0-9a-fA-F]{64}", digest):
                raise ValueError("manifest.json entry {} must contain a valid SHA-256 hash".format(name))
        path = session / name
        if path.stat().st_size != size:
            raise ValueError("Manifest size mismatch for {}".format(name))
        if digest is not None:
            actual = hashlib.sha256()
            with path.open("rb") as source:
                for block in iter(lambda: source.read(1024 * 1024), b""):
                    actual.update(block)
            if actual.hexdigest().lower() != digest.lower():
                raise ValueError("Manifest SHA-256 mismatch for {}".format(name))
        verified.append({"file": name, "bytes": size, "sha256_verified": digest is not None})
    return {"verified": True, "files": verified,
            "note": "Files without a recorded SHA-256 hash were checked by size only."}


def _write_assessment(path, assessment):
    serialized = json.dumps(assessment, indent=2, ensure_ascii=False, allow_nan=False) + "\n"
    # Atomic replacement targets the sidecar entry itself, never a symlink's
    # destination; a failed write leaves any earlier assessment intact.
    descriptor, temporary = tempfile.mkstemp(prefix=".assessment-", suffix=".json", dir=str(path.parent))
    temporary = Path(temporary)
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8") as destination:
            destination.write(serialized)
        os.replace(str(temporary), str(path))
    finally:
        if temporary.exists():
            temporary.unlink()


def review_session(session, assessment_path=None, check_manifest=False):
    """Return report.html after all supplied evidence passes validation."""
    session = Path(session).expanduser().resolve()
    verification = verify_manifest(session) if check_manifest else None
    analysis = analyze_session(session)
    if verification is not None:
        analysis["manifest_verification"] = verification
    if assessment_path is not None:
        with Path(assessment_path).expanduser().open("r", encoding="utf-8-sig") as source:
            assessment = json.load(source)
        accepted = validate_coach_assessment(assessment, analysis)
        _write_assessment(session / "coach-assessment.json", accepted)
    return generate_review(session, analysis=analysis)


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("session", type=Path, help="Directory containing session.json, frames.csv and mouse.csv")
    parser.add_argument("--assessment", type=Path, help="Assistant-authored JSON to validate and save as coach-assessment.json")
    parser.add_argument("--verify-manifest", action="store_true", help="Verify original file sizes/hashes before creating report files")
    args = parser.parse_args(argv)
    try:
        report = review_session(args.session, assessment_path=args.assessment, check_manifest=args.verify_manifest)
    except (OSError, ValueError, TypeError, OverflowError) as error:
        parser.exit(2, "error: {}\n".format(error))
    print(report)
    if args.verify_manifest:
        print("Manifest verified; files lacking a recorded SHA-256 hash were checked by size only.")
    return 0


if __name__ == "__main__":
    sys.exit(main())

scripts/export_evidence.py

"""Export actual recorded frames for offline visual review (Python 3.8+).

Requires FFmpeg on PATH or --ffmpeg. This script never captures a desktop,
reads live input, accesses a game, or calls an AI service. Click candidates
are input events, not confirmed shots, and right-held state is not ADS.
"""

import argparse
import csv
import json
import math
import os
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile


class EvidenceError(RuntimeError):
    """An export could not safely produce a complete evidence index."""


def _read_json(path):
    try:
        with path.open("r", encoding="utf-8-sig") as source:
            value = json.load(source)
    except (OSError, ValueError) as error:
        raise EvidenceError("{}: {}".format(path.name, error)) from error
    if not isinstance(value, dict):
        raise EvidenceError("{} must contain a JSON object".format(path.name))
    return value


def _nonnegative_integer(value, name):
    if isinstance(value, bool) or not isinstance(value, int) or value < 0:
        raise EvidenceError("{} must be a nonnegative integer".format(name))
    return value


def _read_frames(path):
    frames = []
    try:
        with path.open("r", encoding="utf-8-sig", newline="") as source:
            reader = csv.DictReader(source)
            required = {"frame_index", "t_start_ns", "t_end_ns"}
            if not required.issubset(reader.fieldnames or []):
                raise EvidenceError("frames.csv must contain {}".format(", ".join(sorted(required))))
            for line, row in enumerate(reader, 2):
                try:
                    frame = {key: int(row[key]) for key in required}
                except (TypeError, ValueError) as error:
                    raise EvidenceError("frames.csv line {} has invalid integers".format(line)) from error
                if frame["frame_index"] != len(frames):
                    raise EvidenceError("frames.csv line {}: frame indices must be consecutive from zero; "
                                        "cannot verify video correspondence".format(line))
                start, end = frame["t_start_ns"], frame["t_end_ns"]
                if start < 0 or end < start:
                    raise EvidenceError("frames.csv line {} has an invalid capture bracket".format(line))
                frame["t_mid_ns"] = (start + end) // 2
                if frames and frame["t_mid_ns"] < frames[-1]["t_mid_ns"]:
                    raise EvidenceError("frames.csv capture midpoints are out of order at line {}".format(line))
                frames.append(frame)
    except OSError as error:
        raise EvidenceError("frames.csv: {}".format(error)) from error
    if not frames:
        raise EvidenceError("frames.csv contains no recorded frames")
    return frames


def _evenly_spaced(items, count):
    """Sample ordered positions, including both ends when count is at least two."""
    count = min(count, len(items))
    if not count:
        return []
    if count == 1:
        return [items[(len(items) - 1) // 2]]
    # Integer nearest-position rounding avoids floating-point index drift.
    return [items[(i * (len(items) - 1) + (count - 1) // 2) // (count - 1)]
            for i in range(count)]


def _load_candidates(path, session_id, fps, frames):
    analysis = _read_json(path)
    if analysis.get("schema") != "aim-observer/review-v1":
        raise EvidenceError("analysis.json must use aim-observer/review-v1; run review_session.py first")
    if analysis.get("session_id") != session_id:
        raise EvidenceError("analysis.json belongs to a different session; rebuild the analysis")
    if isinstance(analysis.get("fps"), bool) or analysis.get("fps") != fps:
        raise EvidenceError("analysis.json FPS differs from session.json; rebuild the analysis")
    candidates = analysis.get("candidates")
    if not isinstance(candidates, list):
        raise EvidenceError("analysis.json candidates must be a list")
    seen = set()
    for candidate in candidates:
        if not isinstance(candidate, dict):
            raise EvidenceError("Every analysis candidate must be an object")
        candidate_id = candidate.get("id")
        if not isinstance(candidate_id, str) or not candidate_id or candidate_id in seen:
            raise EvidenceError("Analysis candidate IDs must be nonempty and unique")
        seen.add(candidate_id)
        _nonnegative_integer(candidate.get("t_ns"), "{} t_ns".format(candidate_id))
        mapped = candidate.get("mapped_frame")
        if not isinstance(mapped, dict):
            raise EvidenceError("{} has no mapped_frame; rebuild the analysis".format(candidate_id))
        index = _nonnegative_integer(mapped.get("frame_index"), "{} mapped frame_index".format(candidate_id))
        if index >= len(frames):
            raise EvidenceError("{} maps to an unrecorded frame; rebuild the analysis".format(candidate_id))
        for key in ("t_start_ns", "t_end_ns", "t_mid_ns"):
            if key in mapped and mapped[key] != frames[index][key]:
                raise EvidenceError("{} mapped {} differs from frames.csv; rebuild the analysis".format(
                    candidate_id, key))
    return sorted(candidates, key=lambda candidate: (candidate["t_ns"], candidate["id"]))


def _find_ffmpeg(value):
    executable = str(Path(value).expanduser()) if value is not None else "ffmpeg"
    located = shutil.which(executable)
    if located:
        return located
    if value is not None and Path(executable).is_file():
        return str(Path(executable).resolve())
    raise EvidenceError("FFmpeg was not found. Install FFmpeg on PATH or pass --ffmpeg /path/to/ffmpeg")


def _extract_frames(executable, video, indices, staging):
    # A script keeps large selections below Windows' command-line length limit.
    filter_path = staging / "select-frames.txt"
    filter_path.write_text("select=" + "+".join("eq(n\\,{})".format(index) for index in indices),
                           encoding="utf-8")
    pattern = staging / "decoded-%09d.png"
    command = [executable, "-hide_banner", "-loglevel", "error", "-nostdin", "-n",
               "-i", str(video), "-map", "0:v:0", "-an", "-sn", "-dn",
               "-filter_script:v", str(filter_path), "-vsync", "0", "-threads", "1",
               "-frames:v", str(len(indices)), "-start_number", "0", str(pattern)]
    try:
        result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                universal_newlines=True, encoding="utf-8", errors="replace",
                                creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
    except OSError as error:
        raise EvidenceError("Could not start FFmpeg: {}".format(error)) from error
    if result.returncode:
        detail = (result.stderr or result.stdout or "no diagnostic output").strip()[-4000:]
        raise EvidenceError("FFmpeg failed (exit {}): {}".format(result.returncode, detail))
    files = sorted(staging.glob("decoded-*.png"))
    expected = [staging / "decoded-{:09d}.png".format(position) for position in range(len(indices))]
    if files != expected or any(path.stat().st_size == 0 for path in files):
        raise EvidenceError("FFmpeg produced {} images for {} selected frames. The video may be truncated "
                            "or inconsistent with frames.csv; no evidence index was written".format(
                                len(files), len(indices)))
    for position, index in enumerate(indices):
        expected[position].rename(staging / "frame-{:09d}.png".format(index))


def _source_path(path, output):
    try:
        return Path(os.path.relpath(str(path), str(output))).as_posix()
    except ValueError:  # Different drives on Windows cannot have a relative path.
        return path.as_posix()


def export_evidence(session, output=None, click_ids=None, overview_count=12,
                    before_frames=6, after_frames=6, click_count=6,
                    overview_only=False, ffmpeg=None):
    """Export unique PNGs and return evidence.json, refusing nonempty output dirs.

    Selected clicks use analysis.json's mapped frame index. Neighbors are
    consecutive recorded frames, even across capture gaps; their actual timing
    is retained. Overview and default click samples span the ordered recording,
    not the first N items. No source recording or analysis file is modified.
    """
    for name, value in (("overview_count", overview_count), ("before_frames", before_frames),
                        ("after_frames", after_frames), ("click_count", click_count)):
        _nonnegative_integer(value, name)
    if isinstance(click_ids, str):
        raise EvidenceError("click_ids must be a sequence of IDs, not one string")
    requested_ids = list(dict.fromkeys(click_ids or []))
    if any(not isinstance(value, str) or n...