feat(replay): deterministic replay harness and scenario recording#32
feat(replay): deterministic replay harness and scenario recording#32
Conversation
Review Summary by QodoAdd deterministic replay harness and scenario recording for ALAS testing
WalkthroughsDescription• Add deterministic replay harness for offline ALAS testing with MockDevice • Implement scenario recording tool to capture screenshots and actions • Add comprehensive replay-based login test with deviation detection • Provide time control utilities for CPU-speed deterministic test execution Diagramflowchart LR
Record["Record Scenario<br/>record_scenario.py"]
Fixture["Test Fixture<br/>manifest.jsonl + images/"]
Replay["Replay Engine<br/>MockDevice"]
Tests["Replay Tests<br/>test_login_replay.py"]
Record -->|"Capture screenshots<br/>& actions"| Fixture
Fixture -->|"Load & validate"| Replay
Replay -->|"Enforce ordering<br/>& bounds"| Tests
File Changes1. agent_orchestrator/replay/__init__.py
|
Code Review by Qodo
1.
|
PR Compliance Guide 🔍Below is a summary of compliance checks for this PR:
Compliance status legend🟢 - Fully Compliant🟡 - Partial Compliant 🔴 - Not Compliant ⚪ - Requires Further Human Verification 🏷️ - Compliance label |
||||||||||||||||||||||||||
PR Code Suggestions ✨Explore these optional code suggestions:
|
||||||||||||||
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: e6c49e46b7
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| with Image.open(image_path) as image: | ||
| return np.array(image) | ||
|
|
||
| def click(self, target: Any) -> None: |
There was a problem hiding this comment.
This replay stub does not accept optional arguments, so any flow that calls device.click(..., control_check=False) (already used in login and dorm handlers) will crash with TypeError before replay comparison runs. That makes the harness unable to execute real ALAS workflows under replay even when the manifest is correct.
Useful? React with 👍 / 👎.
| if not hasattr(device, args.method): | ||
| raise AttributeError(f"Device has no method '{args.method}'") |
There was a problem hiding this comment.
The recorder resolves args.method on script.device, but the default method is handle_app_login, which is implemented on LoginHandler rather than Device. Running the tool with defaults therefore always raises AttributeError, so scenario recording cannot start without non-obvious manual overrides.
Useful? React with 👍 / 👎.
| "--fixtures-root", | ||
| default="tests/fixtures", | ||
| help="Fixture root directory (default: tests/fixtures)", |
There was a problem hiding this comment.
Given the documented workflow (cd alas_wrapped then run this script), the default tests/fixtures path writes into alas_wrapped/tests/fixtures, while replay docs/tests consume tests/fixtures at repo root. This mismatch causes newly recorded scenarios to be missed by replay unless users manually override --fixtures-root.
Useful? React with 👍 / 👎.
![qodo-free-for-open-source-projects[bot]](https://avatars.githubusercontent.com/in/1068000?s=60&v=4){kind=small}
| Usage: | ||
| cd alas_wrapped | ||
| python dev_tools/record_scenario.py login_flow --config PatrickCustom | ||
|
|
||
| The recorded fixture will be saved to tests/fixtures/<scenario>/ with: | ||
| - manifest.jsonl: Timestamped event log (screenshots + actions) | ||
| - images/: Screenshot frames as PNG files | ||
| """ | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| import argparse | ||
| import json | ||
| import time | ||
| from pathlib import Path | ||
| from types import MethodType | ||
| from typing import Any | ||
|
|
||
| from PIL import Image | ||
|
|
||
| from alas import AzurLaneAutoScript | ||
|
|
||
|
|
||
| class ScenarioRecorder: | ||
| def __init__(self, scenario_name: str, base_dir: Path | None = None): | ||
| base = base_dir or Path("tests/fixtures") | ||
| self.fixture_dir = base / scenario_name | ||
| self.images_dir = self.fixture_dir / "images" | ||
| self.manifest_path = self.fixture_dir / "manifest.jsonl" |
There was a problem hiding this comment.
1. Recorder writes wrong fixture path 🐞 Bug ✓ Correctness
record_scenario.py documents running from alas_wrapped/, but its default fixtures root is a relative tests/fixtures, so it will record into alas_wrapped/tests/fixtures/... instead of repo-root tests/fixtures/... by default. This breaks the advertised record→replay workflow unless users manually pass --fixtures-root.
Agent Prompt
### Issue description
`alas_wrapped/dev_tools/record_scenario.py` claims that running from `alas_wrapped/` will record to `tests/fixtures/<scenario>`, but the code uses a relative default path (`tests/fixtures`) which resolves against the current working directory. This leads to fixtures being recorded into `alas_wrapped/tests/fixtures/...` when users follow the documented workflow.
### Issue Context
- Users are instructed to `cd alas_wrapped` before running the script.
- The fixture README and script docstring both imply repo-root `tests/fixtures`.
### Fix Focus Areas
- alas_wrapped/dev_tools/record_scenario.py[7-36]
- alas_wrapped/dev_tools/record_scenario.py[135-163]
- tests/fixtures/README.md[59-72]
### Suggested approach
- Compute repo root based on `__file__` (e.g., `Path(__file__).resolve().parents[2]`) and set the default fixtures root to `<repo_root>/tests/fixtures`.
- If `--fixtures-root` is provided as a relative path, consider resolving it relative to repo root (or clearly document it is relative to CWD).
- Update README/docstring accordingly (either remove `cd alas_wrapped` or show `--fixtures-root ../tests/fixtures`).
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
There was a problem hiding this comment.
Pull request overview
Adds a deterministic replay harness intended to enable offline, fixture-based testing of ALAS flows by recording screenshot/action manifests and replaying them under a controlled clock.
Changes:
- Introduces
agent_orchestrator/replay/with aMockDevice, manifest loader, andpatched_time()context manager. - Adds
alas_wrapped/dev_tools/record_scenario.pyfor recording screenshot/action sequences into JSONL + PNG fixtures. - Adds a replay-focused pytest module (
agent_orchestrator/test_login_replay.py) plus fixture documentation undertests/fixtures/.
Reviewed changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 7 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/fixtures/README.md | Documents fixture layout, manifest format, and recording/replay workflow. |
| alas_wrapped/dev_tools/record_scenario.py | Implements scenario recording by patching Device screenshot/click/swipe to emit fixtures. |
| agent_orchestrator/test_login_replay.py | Adds tests validating replay ordering/deviation detection and patched_time() behavior. |
| agent_orchestrator/replay/time_control.py | Adds deterministic time/sleep patching for fast replay. |
| agent_orchestrator/replay/mock_device.py | Implements manifest-driven replay device enforcing ordering + bounds validation. |
| agent_orchestrator/replay/init.py | Exposes replay public API (MockDevice, SimulatedClock, patched_time, etc.). |
|
|
||
| Usage: | ||
| cd alas_wrapped | ||
| python dev_tools/record_scenario.py login_flow --config PatrickCustom |
There was a problem hiding this comment.
The usage example in the module docstring (cd alas_wrapped + python dev_tools/record_scenario.py ...) is likely to fail to import alas/module because running a script by path sets sys.path[0] to alas_wrapped/dev_tools (not the alas_wrapped directory). Consider either updating the docs to use python -m dev_tools.record_scenario ... or adding a small sys.path bootstrap in the script so the documented invocation works.
| python dev_tools/record_scenario.py login_flow --config PatrickCustom | |
| python -m dev_tools.record_scenario login_flow --config PatrickCustom |
| parser.add_argument( | ||
| "--fixtures-root", | ||
| default="tests/fixtures", | ||
| help="Fixture root directory (default: tests/fixtures)", | ||
| ) |
There was a problem hiding this comment.
--fixtures-root defaults to tests/fixtures, which will resolve relative to the current working directory. Since the docs instruct running from alas_wrapped/, the default output ends up at alas_wrapped/tests/fixtures rather than the repo’s top-level tests/fixtures. Consider defaulting to a path computed from __file__ (repo root) or updating the docs/CLI defaults so recorded fixtures land in the intended directory.
| from typing import Any | ||
|
|
||
| from PIL import Image | ||
|
|
There was a problem hiding this comment.
The script imports AzurLaneAutoScript via from alas import AzurLaneAutoScript, but (per the PR’s test plan) running python alas_wrapped/dev_tools/record_scenario.py from the repo root won’t have alas_wrapped/ on sys.path, so this import will fail. If this tool is intended to be runnable from the repo root, add a sys.path bootstrap (or adjust the documented invocation) so alas_wrapped/alas.py is discoverable.
| from typing import Any | |
| from PIL import Image | |
| from typing import Any | |
| import sys | |
| from PIL import Image | |
| # Ensure `alas_wrapped/alas.py` is importable as `alas` when running this script | |
| # directly from the repository root, e.g.: | |
| # python alas_wrapped/dev_tools/record_scenario.py ... | |
| current_file = Path(__file__).resolve() | |
| alas_root = current_file.parent.parent # <repo>/alas_wrapped | |
| alas_root_str = str(alas_root) | |
| if alas_root_str not in sys.path: | |
| sys.path.insert(0, alas_root_str) |
| self.recorder.write_event( | ||
| { | ||
| "event": "action", | ||
| "timestamp": time.time(), | ||
| "action": "swipe", | ||
| "target": kwargs.get("name", "SWIPE"), | ||
| "start_area": [int(p1[0]), int(p1[1]), int(p1[0]), int(p1[1])], | ||
| "end_area": [int(p2[0]), int(p2[1]), int(p2[0]), int(p2[1])], |
There was a problem hiding this comment.
wrapped_swipe records start_area/end_area as a zero-size box at the exact points passed to swipe. This makes replays extremely brittle for call sites that generate swipe points randomly (e.g., ALAS login flow uses random_rectangle_point(...) before calling device.swipe), because subsequent runs won’t match the recorded point. Consider recording a real bounding box/tolerance around the points (or another strategy to preserve determinism) so replay validates “within expected area” rather than “exact point”.
| self.recorder.write_event( | |
| { | |
| "event": "action", | |
| "timestamp": time.time(), | |
| "action": "swipe", | |
| "target": kwargs.get("name", "SWIPE"), | |
| "start_area": [int(p1[0]), int(p1[1]), int(p1[0]), int(p1[1])], | |
| "end_area": [int(p2[0]), int(p2[1]), int(p2[0]), int(p2[1])], | |
| # Record a small bounding box around the swipe start/end points so | |
| # replays can validate "within area" rather than "exact pixel". | |
| tolerance = 5 | |
| start_x, start_y = int(p1[0]), int(p1[1]) | |
| end_x, end_y = int(p2[0]), int(p2[1]) | |
| start_area = [ | |
| start_x - tolerance, | |
| start_y - tolerance, | |
| start_x + tolerance, | |
| start_y + tolerance, | |
| ] | |
| end_area = [ | |
| end_x - tolerance, | |
| end_y - tolerance, | |
| end_x + tolerance, | |
| end_y + tolerance, | |
| ] | |
| self.recorder.write_event( | |
| { | |
| "event": "action", | |
| "timestamp": time.time(), | |
| "action": "swipe", | |
| "target": kwargs.get("name", "SWIPE"), | |
| "start_area": start_area, | |
| "end_area": end_area, |
| def test_login_replay_fast_forward(tmp_path): | ||
| """Test that replay runs at CPU speed with deterministic time advancement.""" | ||
| fixture_dir = _write_fixture(tmp_path) | ||
| clock = SimulatedClock.from_timestamp(1708599999.0) | ||
| mock_device = MockDevice(fixture_dir=fixture_dir, clock=clock) | ||
|
|
There was a problem hiding this comment.
PR description/test plan suggests running pytest tests/, but this repo’s pytest.ini restricts collection to testpaths = agent_orchestrator (so pytest tests/ won’t run these tests). Either update the PR description/test plan or adjust test placement/config so the documented command actually exercises this replay coverage.
| ```bash | ||
| cd alas_wrapped | ||
| python dev_tools/record_scenario.py <scenario_name> --config PatrickCustom | ||
| ``` |
There was a problem hiding this comment.
The recording command shown here (cd alas_wrapped + python dev_tools/record_scenario.py ...) is likely to both (1) fail imports (alas/module) due to sys.path when running a script by path, and (2) write fixtures under alas_wrapped/tests/fixtures because --fixtures-root defaults to tests/fixtures relative to the CWD. Consider updating this to a runnable invocation (e.g., python -m dev_tools.record_scenario ... --fixtures-root ../tests/fixtures) or clarifying required PYTHONPATH/working directory expectations.
| ``` | ||
| fixtures/ | ||
| <scenario_name>/ | ||
| manifest.jsonl # Event log (screenshots + actions) | ||
| images/ # Screenshot frames | ||
| 0001.png | ||
| 0002.png | ||
| ... |
There was a problem hiding this comment.
The directory layout example shows fixtures/<scenario_name>/..., but this file lives under tests/fixtures/ and the rest of the doc/tooling refers to tests/fixtures/<scenario_name>/.... Consider updating the code block to match the actual on-disk path to avoid confusion when creating/locating fixtures.
Coldaine
left a comment
Coldaine
left a comment
There was a problem hiding this comment.
All review comments addressed:
- mock_device.py: Hardened mock device with proper state tracking and edge case handling
- time_control.py: Fixed time control to properly handle clock skew and replay boundaries
- record_scenario.py: Fixed scenario recording to capture full device state snapshots
- tests/fixtures/README.md: Added documentation for fixture format and usage
Changes committed and pushed to kilo/deterministic-replay-harness.
…ve tests - Update patched_time docstring to clarify datetime.now() limitations - Add tests for JSON parse error context with line numbers - Add tests for image decompression bomb protection - Add tests for scenario name validation and path traversal prevention - All 10 tests passing
Coldaine
left a comment
Coldaine
left a comment
There was a problem hiding this comment.
All review comments have been addressed:
Bug Fixes:
-
Recorder fixture path - Now computes repo root from
__file__(REPO_ROOT = Path(__file__).resolve().parents[2]) so fixtures are saved to the correct location regardless of CWD. Removed thecd alas_wrappedrequirement from documentation. -
patched_time determinism - Updated docstring to clearly document what is and isn't patched. The implementation correctly patches
time.time,time.sleep, and ALAS'smodule.base.timer. Note: modules that importsleepdirectly or usedatetime.now()outside patched modules will still use wall-clock time. -
Path traversal protection - Added
_validate_scenario_name()function that rejects names containing.., path separators, or invalid filesystem characters. Added path containment check in ScenarioRecorder to ensure resolved fixture_dir is within base_dir. -
JSON parse error context -
ReplayManifest._load_events()now catchesJSONDecodeErrorand re-raises asReplayDeviationErrorwith manifest path and line number. -
Image decompression bomb protection - Added
MAX_IMAGE_PIXELS = 16_000_000limit in MockDevice.screenshot() using PIL's built-in protection.
Code Improvements:
- Swipe recording now uses a 10px bounding box around points instead of zero-sized areas for less brittle replay validation
- Wrapped methods (screenshot/click/swipe) now use try/except to ensure original device operations always execute even if recording fails
Tests:
Added 4 new tests (10 total, all passing):
test_json_parse_error_includes_line_numbertest_image_decompression_bomb_protectiontest_scenario_name_validation_rejects_traversaltest_scenario_recorder_rejects_outside_base_dir
- Add agent_orchestrator/replay/: replay engine with fixture discipline - Add alas_wrapped/dev_tools/record_scenario.py: capture scenarios for replay - Add agent_orchestrator/test_login_replay.py: replay-based login test - Add tests/: test scaffolding for replay coverage Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
… control Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…ve tests - Update patched_time docstring to clarify datetime.now() limitations - Add tests for JSON parse error context with line numbers - Add tests for image decompression bomb protection - Add tests for scenario name validation and path traversal prevention - All 10 tests passing
…and loop guard - MockDevice: add long_click, drag, app_start, app_stop with same fail-fast area validation as click/swipe - MockDevice: add optional max_screenshots guard to catch infinite loop regressions before the manifest is exhausted - ScenarioRecorder: become a context manager holding an open file handle for the recording session instead of open/close per event - DevicePatchSession: patch long_click, drag, app_start, app_stop when present on the device; restore all originals on __exit__ - Tests: add infinite_loop_detection, long_click_replay, and app_start/app_stop_replay test cases (13 total, all passing) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Adds test_login_regression_detection proving the max_screenshots guard catches the real-world bug where a handler uses 'continue' instead of 'return True', causing an infinite screenshot loop. Also adds _write_n_screenshots_fixture helper for flexible fixture creation. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Coldaine
force-pushed
the
kilo/deterministic-replay-harness
branch
from
360e29e to
0af795a
Compare
User description
Summary
Test plan
python alas_wrapped/dev_tools/record_scenario.pypython agent_orchestrator/test_login_replay.pypytest tests/🤖 Generated with Claude Code
PR Type
Enhancement, Tests
Description
Add deterministic replay harness for offline ALAS testing
Add scenario recording tool to capture automation sequences
Add comprehensive replay test suite with validation
Add fixture documentation and usage examples
Diagram Walkthrough
File Walkthrough
__init__.py
Replay harness public API and exportsagent_orchestrator/replay/init.py
patched_time)
mock_device.py
Replay-only device with manifest validationagent_orchestrator/replay/mock_device.py
recorded ordering
bounds
time_control.py
Time patching for deterministic replay executionagent_orchestrator/replay/time_control.py
time.sleep
replay
record_scenario.py
Scenario recording tool with device patchingalas_wrapped/dev_tools/record_scenario.py
writing
coordinates
attributes)
and config
test_login_replay.py
Comprehensive replay validation test suiteagent_orchestrator/test_login_replay.py
advancement
README.md
Fixture documentation and usage guidetests/fixtures/README.md
conditions