feat(tutorial): add guided human recording and paced replay

README.md

@@ -57,6 +57,20 @@ application itself never calls, so the screen cannot influence it. It ends
 `VERIFIED` with zero model calls, and writes a shareable `receipt.png` /
 `receipt.json` beside the run.
 
+For a live walkthrough, perform the demonstration yourself and then watch the
+compiled replay at a visible pace:
+
+```bash
+openadapt-flow tutorial --guided
+```
+
+The recording browser closes after OpenAdapt observes the saved record through
+the separate read-only interface. OpenAdapt then compiles, certifies, and
+replays what you demonstrated. If you prefer a fully automatic presentation, use
+`openadapt-flow tutorial --headed --presentation-delay 1`. The delay applies
+only to this bundled tutorial. The ordinary `tutorial`, `replay`, and `run`
+paths keep their normal execution speed.
+
 That receipt is generated from a closed allow-list — outcomes, counts, digests,
 and validated package versions — so it can carry no screenshot, OCR text,
 typed value, parameter, URL, hostname, coordinate, operator text, or free-form

docs/EXECUTION_PROFILES.md

@@ -71,6 +71,32 @@ The tutorial also omits the login screen: typing a credential into a recording
 produces an artifact whose plaintext value is a secret carrier for no
 evidentiary gain, and the demonstration proves exactly as much without it.
 
+The fast command remains suitable for a quickstart and CI:
+
+```bash
+openadapt-flow tutorial
+```
+
+For a guided presentation, the operator can perform the same demonstration in
+the browser. The interactive recorder retains the separate read-only
+system-of-record snapshots needed for effect mining. The recording browser
+closes after OpenAdapt observes the saved record through that separate
+read-only interface. OpenAdapt then compiles, certifies, and replays the result
+at a visible pace:
+
+```bash
+openadapt-flow tutorial --guided
+```
+
+The automatic alternative is:
+
+```bash
+openadapt-flow tutorial --headed --presentation-delay 1
+```
+
+The delay is bounded to the bundled tutorial. It does not change ordinary
+`replay`, `run`, or production runtime timing.
+
 The result is checkable rather than asserted: run the same bundle against an
 injected backend fault and it does not verify. Both injected faults terminate
 `HALTED` / `RECONCILIATION_REQUIRED`, and both are pinned in

openadapt_flow/__main__.py

@@ -925,6 +925,7 @@ def _cmd_tutorial(args: argparse.Namespace) -> int:
     from datetime import datetime, timezone
 
     from openadapt_flow.tutorial import (
+        GUIDED_PRESENTATION_DELAY_S,
         TUTORIAL_WORKFLOW_NAME,
         TutorialError,
         run_tutorial,
@@ -936,23 +937,33 @@ def _cmd_tutorial(args: argparse.Namespace) -> int:
         else Path("tutorials")
         / ("tutorial-" + datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S"))
     )
+    interactive_record = bool(args.interactive_record or args.guided)
+    presentation_delay_s = args.presentation_delay
+    if presentation_delay_s is None:
+        presentation_delay_s = GUIDED_PRESENTATION_DELAY_S if args.guided else 0.0
     try:
         result = run_tutorial(
             out,
-            headed=args.headed,
+            headed=bool(args.headed or args.guided),
             name=args.name or TUTORIAL_WORKFLOW_NAME,
             emit_receipt=not args.no_receipt,
+            interactive_record=interactive_record,
+            presentation_delay_s=presentation_delay_s,
             echo=print,
         )
     except TutorialError as e:
         print(f"\nTutorial REFUSED: {e}")
         return 2
 
     print(f"\n{result.execution_outcome}: {result.run_dir / 'REPORT.md'}")
-    print(
-        f"  transaction     {result.transaction_outcome} "
-        f"(billable: {'yes' if result.transaction_billable else 'no'})"
-    )
+    print(f"  transaction     {result.transaction_outcome}")
+    metering_class = "billable" if result.transaction_billable else "not billable"
+    local_charge = (
+        "reported"
+        if result.reported_to_metering
+        else "this local tutorial was not reported or charged"
+    )
+    print(f"  metering class  {metering_class} ({local_charge})")
     print(f"  profile         {result.execution_profile}")
     print(f"  model calls     {result.model_calls}")
     print(
@@ -3647,6 +3658,32 @@ def build_parser() -> argparse.ArgumentParser:
         help="Workflow name for the compiled bundle (default: local-quickstart)",
     )
     p.add_argument("--headed", action="store_true", help="Run the browser headed")
+    p.add_argument(
+        "--guided",
+        action="store_true",
+        help=(
+            "Presentation mode: record the demonstration yourself, then watch "
+            "a visibly paced governed replay"
+        ),
+    )
+    p.add_argument(
+        "--interactive-record",
+        action="store_true",
+        help=(
+            "Perform the tutorial demonstration yourself in a headed browser; "
+            "the browser closes after the saved record is observed"
+        ),
+    )
+    p.add_argument(
+        "--presentation-delay",
+        type=float,
+        default=None,
+        metavar="SECONDS",
+        help=(
+            "Tutorial-only pause before each scripted recording action and "
+            "replay step (0-5 seconds; guided default: 1)"
+        ),
+    )
     p.add_argument(
         "--no-receipt",
         action="store_true",

openadapt_flow/interactive_recorder.py

@@ -382,6 +382,10 @@ def __init__(
         settle_stable_frames: int = 2,
         settle_interval_s: float = 0.15,
         viewport: tuple[int, int] = (1280, 800),
+        system_of_record_reader: Optional[
+            Callable[[], Optional[list[dict[str, Any]]]]
+        ] = None,
+        stop_when: Optional[Callable[[], bool]] = None,
     ) -> None:
         self._url = url
         self._out_dir = Path(out_dir)
@@ -391,6 +395,12 @@ def __init__(
         self._headless = headless
         self._poll_ms = poll_ms
         self._viewport = viewport
+        # Recording-only, read-only observation. This does not add an effect
+        # verifier to the browser backend or grant the later runtime authority.
+        self._system_of_record_reader = system_of_record_reader
+        # Optional recording completion condition. It is evaluated only after
+        # queued browser events have been persisted, while the page is open.
+        self._stop_when = stop_when
         self._settle = dict(
             settle_timeout_s=settle_timeout_s,
             settle_stable_frames=settle_stable_frames,
@@ -450,18 +460,30 @@ def start(self) -> None:
             pass
         self.backend = PlaywrightBackend(self.page)
         self.recorder = Recorder(
-            self.backend, self._out_dir, app_url=self._url, **self._settle
+            self.backend,
+            self._out_dir,
+            app_url=self._url,
+            system_of_record_reader=self._system_of_record_reader,
+            **self._settle,
         )
         self._last_frame = self.recorder._wait_settled()
         self._last_structural = self._structural_state()
 
     def run(self) -> Path:
-        """Human loop: pump until the user stops (Ctrl-C / closes the window),
-        then flush and finish."""
+        """Pump until completion, an operator stop, or a closed window."""
+        if self._stop_when is None:
+            finish_instruction = (
+                "Press Ctrl-C here (or close the browser window) to finish."
+            )
+        else:
+            finish_instruction = (
+                "Complete the workflow. Recording stops automatically after "
+                "the configured result is observed. Press Ctrl-C only to stop early."
+            )
         print(
             f"Recording {self._url}\n"
             "  Perform your workflow in the browser window.\n"
-            "  Press Ctrl-C here (or close the browser window) to finish."
+            f"  {finish_instruction}"
         )
         try:
             while not self.done:
@@ -502,6 +524,8 @@ def pump(self) -> bool:
         return self._pump()
 
     def _pump(self) -> bool:
+        if self.done:
+            return False
         try:
             self.page.wait_for_timeout(self._poll_ms)
         except Exception:
@@ -515,10 +539,10 @@ def _pump(self) -> bool:
             # is NOT idle-flushed (a mid-word pause must not split it) — it
             # flushes on the next boundary event or at finish().
             self._flush_scroll()
-            return True
+            return not self._stop_condition_reached()
         for ev in batch:
             self._process(ev)
-        return True
+        return not self._stop_condition_reached()
 
     def _process(self, ev: dict[str, Any]) -> None:
         kind = ev.get("kind")
@@ -735,8 +759,26 @@ def _structural_state(self) -> dict[str, Any]:
                 value = None
             if value is not None:
                 state[key] = value
+        if self._system_of_record_reader is not None:
+            try:
+                records = self._system_of_record_reader()
+            except Exception:
+                records = None
+            if records is not None:
+                state["sor"] = records
         return state
 
+    def _stop_condition_reached(self) -> bool:
+        if self._stop_when is None:
+            return False
+        try:
+            done = bool(self._stop_when())
+        except Exception:
+            done = False
+        if done:
+            self.done = True
+        return done
+
 
 def record_interactive(
     url: str,
@@ -747,6 +789,10 @@ def record_interactive(
     identifier_fields: tuple[str, ...] = (),
     headless: bool = False,
     script: Optional[Callable[[Any, Callable[[], None]], None]] = None,
+    system_of_record_reader: Optional[
+        Callable[[], Optional[list[dict[str, Any]]]]
+    ] = None,
+    stop_when: Optional[Callable[[], bool]] = None,
     **kwargs: Any,
 ) -> Path:
     """Record a live demonstration the user drives against ``url``.
@@ -773,6 +819,13 @@ def record_interactive(
             a human recording is headed).
         script: Test hook — ``script(page, pump)`` drives synthetic input and
             pumps the loop; when given, the human wait loop is skipped.
+        system_of_record_reader: Optional read-only observation of the
+            authoritative records. The recorder retains the observed
+            before/after state so compilation can propose effect contracts.
+            This observer is recording-only and does not become a runtime
+            verifier.
+        stop_when: Optional recording completion condition. It is evaluated
+            after queued events are persisted and before the browser closes.
 
     Returns:
         The recording directory.
@@ -784,6 +837,8 @@ def record_interactive(
         param_fields=param_fields,
         identifier_fields=identifier_fields,
         headless=headless,
+        system_of_record_reader=system_of_record_reader,
+        stop_when=stop_when,
         **kwargs,
     )
     session.start()