Policy-driven scan feedback (abort / restart / kill)

[HarlanHeilman]

Policy-driven scan feedback (abort / restart / kill)

Summary

Add optional user-defined policies to the scan executor so that feedback runs on a configurable interval (e.g. every N points). Policies receive the current result history and can request abort, restart with a fix, or kill. The scan plan remains user-designed; policies only act when their conditions indicate something has gone wrong.

Background

The desired automation flow is:

1. Human defines initial positions and scan/experiment design (scan plan).
2. Motors move to positions; data is collected (AI, area detector).
3. Data is processed and passed to a program/function/agent.
4. The agent decides if data quality is acceptable and whether to recollect (different motor positions or exposure) or continue.
5. The agent can determine the next point or request stop/restart.

This issue focuses on a minimal, plan-first design: the user still designs the scan plan; policies are optional overlays that run on a schedule and only react when conditions fire (abort, restart with fix, or kill).

Proposed behaviour

Policy evaluation interval

• Add a parameter policy_interval: int (default 1) to the scan execution API.
• Policies are evaluated only every policy_interval completed points (e.g. every 10 points).
• Optionally evaluate after the first point so the first check is not delayed until 10 points.

User-defined policies (optional)

Three optional callables, each taking history: list[ScanResult] and returning a decision. If not provided, that check is skipped (current behaviour preserved).

Policy	Signature	Meaning
abort_policy	(history: list[ScanResult]) -> bool	If True, stop the scan after the current point and return partial results. Run is closed with an aborted status. No restart.
restart_policy	(history: list[ScanResult]) -> RestartDecision	If returns a restart-with-fix decision, apply the fix (e.g. retry with new exposure, insert a point) and continue; otherwise continue with the next plan point.
kill_policy	(history: list[ScanResult]) -> bool	If True, stop immediately and do not attempt any restart. Use for unrecoverable or safety conditions.

RestartDecision and fix types

• RestartDecision: either “no restart” or “restart with fix”.
• Fix variants (to be designed in implementation):
  • Retry last point with a new exposure time.
  • Insert point: execute a given ScanPoint next, then resume the plan.
  • Replace next point: replace the next planned point with a given ScanPoint (e.g. same motors, different exposure).

Execution order

When it is time to evaluate policies (every policy_interval points):

1. Evaluate kill_policy(history); if True, stop and close run (e.g. exit_status="killed" or "failed"), return partial results.
2. Evaluate abort_policy(history); if True, stop and close run (exit_status="aborted"), return partial results.
3. Evaluate restart_policy(history); if result is “restart with fix”, apply the fix (insert/replace point or retry last point), then continue; otherwise continue with the next point in the plan.

API surface

• Extend ScanExecutor.execute_scan() (and/or Beamline.scan_from_dataframe()) with optional keyword arguments:
  • abort_policy: Callable[[list[ScanResult]], bool] | None = None
  • restart_policy: Callable[[list[ScanResult]], RestartDecision] | None = None
  • kill_policy: Callable[[list[ScanResult]], bool] | None = None
  • policy_interval: int = 1
• When all policy arguments are omitted, behaviour is unchanged (no policy checks).

Example usage (conceptual)

def abort_policy(history: list[ScanResult]) -> bool:
    # Stop if user condition met (e.g. beam loss)
    return len(history) > 0 and some_beam_loss_condition(history[-1])

def restart_policy(history: list[ScanResult]) -> RestartDecision:
    # If last point was bad, retry with 2x exposure
    if len(history) < 10:
        return NoRestart()
    last = history[-1]
    if detector.check_exposure(last.image).underexposed:
        return RestartWithFix(RetryLastPoint(exposure_seconds=2 * last.exposure_time))
    return NoRestart()

def kill_policy(history: list[ScanResult]) -> bool:
    # Unrecoverable: e.g. hardware fault
    return any(is_fatal(r) for r in history[-5:])

await bl.scan_from_dataframe(
    df,
    abort_policy=abort_policy,
    restart_policy=restart_policy,
    kill_policy=kill_policy,
    policy_interval=10,
)

Acceptance criteria

• policy_interval is supported; policies are invoked every N completed points (and optionally after the first point).
• abort_policy(history) when returning True stops the scan and returns partial results with run status aborted.
• kill_policy(history) when returning True stops the scan without restart; run status reflects killed/failed.
• restart_policy(history) can return a “restart with fix” (e.g. retry last point with new exposure, or insert point); the executor applies the fix and continues.
• When none of the policies are passed, scan behaviour is unchanged from current implementation.
• Types for RestartDecision and fix variants are defined and documented (e.g. in resonance.api.types or a dedicated module).
• Docs and docstrings describe the three policies, evaluation order, and policy_interval.

Out of scope for this issue

• Fully adaptive scans where the agent is the sole source of the next point (separate feature).
• Per-point callback that runs on every point (this issue is interval-based and “only when something goes wrong”).

References

• Current scan execution: ScanExecutor.execute_scan, Beamline.scan_from_dataframe in src/resonance/api/core/scan.py and beamline.py.
• Result type: ScanResult in src/resonance/api/types.py.
• Detector quality: AreaDetector.check_exposure in src/resonance/api/core/det.py.