senofy/autofirmer-expanded
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Revamp login guards into general per-firm page guards

Browse Source 
Replace the single hardcoded signed-out check with a guards list on each
firm: any guard can trip mid-run via a URL substring (checked on every
step) or a selector probe (checked when a step fails). Each guard defines
recovery steps or a human-wait fallback, a per-run trigger cap, and a
resume policy — 'continue' retries the tripped step, 'restart' rewinds
the run to step 1.

Runner bumped to 0.20.0; the server still ships the legacy
signedOut/authSteps shape for stale runners, and a new runner synthesizes
a guard from a legacy payload.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Brandon Li
2026-09-14 12:50:14 -05:00
co-authored by Claude Fable 5
parent 82b98b9d2d
commit b1dccd7dec
5 changed files with 371 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
clicker/actions.py
+38 -19
View File
@@ -15,7 +15,8 @@ class StepError(Exception):
"""A step refused to run, or ran and could not be verified.
`code` mirrors the CLI's exit codes: 3 = refused before acting,
4 = acted but verification failed, 5 = the session is signed out.
4 = acted but verification failed, 5 = a page guard tripped (login
redirect, blocking modal, and the like).
"""
def __init__(self, message: str, code: int = 3):
@@ -23,33 +24,44 @@ class StepError(Exception):
self.code = code
class SignedOutError(StepError):
"""The tab is on the login page, so the session has expired.
class GuardTripped(StepError):
"""A page guard's condition is on the page — the run cannot proceed past it.
Distinct from an ordinary failure because nothing is wrong with the
automation — it just cannot proceed until a human signs in.
Distinct from an ordinary failure because nothing is wrong with the step:
the page has been taken over by something (a login redirect, an expired-
session modal) and every selector now resolves against the wrong document.
A click or a keystroke aimed at the old page would land on whatever the
guard page happens to render in the same place. The runner decides how to
recover; the CLI just stops with exit code 5.
"""
def __init__(self, url: str):
def __init__(self, guard: dict, url: str):
name = guard.get("label") or guard.get("id") or "page"
super().__init__(
f"signed out — the tab is on {url}. Log in to the site, then run this again.",
f"{name} guard tripped — the tab is on {url}. "
"Deal with the page, then run this again.",
code=5,
)
self.guard = guard
self.url = url
def check_signed_in(found: dict, signed_out: str) -> None:
"""Abort if the page redirected to a login screen.
def check_guards(found: dict, guards: list | None) -> None:
"""Abort if the page has tripped any URL guard.
Worth doing before every action, not just at the start: a session can expire
mid-run, and once it does, every selector resolves against a login form. A
click or a keystroke aimed at the old page would land on whatever that form
happens to render in the same place.
Worth doing before every action, not just at the start: a session can
expire (or a guard page appear) mid-run. Only URL guards are checked here —
the URL rides along with every locate answer, so they are free. Selector
guards cost a locate round trip each and are probed by the runner only once
a step has already failed.
"""
if not signed_out:
if not guards:
return
url = found.get("url") or ""
if signed_out in url:
raise SignedOutError(url)
for guard in guards:
pattern = guard.get("urlContains") or ""
if pattern and pattern in url:
raise GuardTripped(guard, url)
# Display scaling factors that actually exist. Anything else means the two sides
@@ -117,6 +129,7 @@ def perform(
url: str = "",
open_url: str = "",
signed_out: str = "",
guards: list | None = None,
frame: str = "",
text: str | None = None,
clear: bool = False,
@@ -142,6 +155,12 @@ def perform(
rng = rng or random.Random()
say = report or (lambda _m: None)
# `guards` is the runner's list of page guards; `signed_out` is the CLI's
# single --signed-out pattern. Same check either way, so fold them together.
guards = list(guards or [])
if signed_out:
guards.append({"id": "signed-out", "label": "signed out", "urlContains": signed_out})
# `frame` names the <iframe> the element lives in; the extension resolves the
# selector inside it and offsets the coordinates back. Absent for the common
# top-document case, and then locate behaves exactly as before.
@@ -150,7 +169,7 @@ def perform(
found = dash.locate(selector, index, url, timeout, open_url, options=frame_opts)
# Before any check that assumes we are on the real page.
check_signed_in(found, signed_out)
check_guards(found, guards)
# Give a transient overlay a chance to finish leaving before giving up on it.
if found.get("covered") and not force:
@@ -160,7 +179,7 @@ def perform(
while found.get("covered") and time.time() < deadline:
time.sleep(COVERED_POLL_SECONDS)
found = dash.locate(selector, index, url, timeout, open_url, options=frame_opts)
check_signed_in(found, signed_out)
check_guards(found, guards)
if not found.get("covered"):
say("it cleared")
@@ -231,7 +250,7 @@ def perform(
# Typing into a field that never took focus, or that ignores the input, fails
# silently and is otherwise indistinguishable from success.
after = dash.locate(selector, index, url, timeout, open_url, options=frame_opts)
check_signed_in(after, signed_out)
check_guards(after, guards)
got = after.get("value")
if got is None:
say("element exposes no value to verify against")
clicker/runner.py
+206 -99
View File
@@ -34,7 +34,8 @@ HEARTBEAT_SECONDS = 2.0
# Bumped whenever the step vocabulary or the locate protocol changes. Reported in
# the heartbeat so the dashboard can say "restart your runner" instead of letting
# a stale process fail on a step type it has never heard of.
VERSION = "0.19.0"
# 0.20.0: page guards — run-level `guards` replace per-step signedOut/authSteps.
VERSION = "0.20.0"
# Shared with the heartbeat thread: whether a run is currently executing.
_busy = threading.Event()
@@ -57,8 +58,14 @@ def heartbeat_loop(dash: Dashboard, opts, stop: threading.Event) -> None:
stop.wait(HEARTBEAT_SECONDS)
def run_one_step(dash: Dashboard, step: dict, opts, rng, report, run_id: int | None = None) -> None:
"""Dispatch a single step. Raises StepError / DashboardError on failure."""
def run_one_step(dash: Dashboard, step: dict, opts, rng, report,
run_id: int | None = None, guards: list | None = None) -> None:
"""Dispatch a single step. Raises StepError / DashboardError on failure,
GuardTripped when a URL guard's condition shows up on the page.
`guards` is empty for recovery steps: they run while a guard's condition
holds, so checking it again would trip on the very page they exist to fix.
"""
if step["action"] == "wait":
time.sleep(float(step.get("seconds", 1)))
report(f"waited {step.get('seconds', 1)}s")
@@ -72,7 +79,7 @@ def run_one_step(dash: Dashboard, step: dict, opts, rng, report, run_id: int | N
navigate_url=target)
# Navigating to a protected page is where an expired session shows up:
# the site answers with a redirect to its login form.
actions.check_signed_in(landed, step.get("signedOut", "") or "")
actions.check_guards(landed, guards)
report(f"tab is on {landed.get('url', target)}")
elif step["action"] == "waitFor":
@@ -110,6 +117,10 @@ def run_one_step(dash: Dashboard, step: dict, opts, rng, report, run_id: int | N
step.get("urlPattern", "") or opts.url,
opts.timeout, step.get("openUrl", "") or "",
options=loc_opts or None)
# A guard page appearing mid-gate would otherwise burn the whole
# gate timeout before anyone noticed. GuardTripped is not caught
# below, so it escapes the wait as it should.
actions.check_guards(seen, guards)
if by_sight:
# An extension too old for the probe quietly does an ordinary
# locate instead, whose result has no `visible` key — reading
@@ -197,7 +208,7 @@ def run_one_step(dash: Dashboard, step: dict, opts, rng, report, run_id: int | N
index=int(step.get("index", 0)),
url=step.get("urlPattern", "") or opts.url,
open_url=step.get("openUrl", "") or "",
signed_out=step.get("signedOut", "") or "",
guards=guards,
frame=step.get("frame", "") or "",
text=step.get("text"),
clear=bool(step.get("clear")),
@@ -224,73 +235,195 @@ def run_status(dash: Dashboard, run_id: int) -> str | None:
return None
def wait_for_sign_in(dash: Dashboard, step: dict, run_id: int, opts, deadline: float) -> bool:
"""Block until the tab leaves the login page.
def guard_present(dash: Dashboard, guard: dict, step: dict, opts,
probe_timeout: float = 6.0) -> bool | None:
"""Is the guard's condition on the page right now? None = could not tell.
The runner does not log in — entering credentials is the human's job, and a
keystroke aimed at the wrong field is not something to risk automating. All
this does is watch the tab and pick the run back up once you are through.
A URL guard reads the tab's URL off a cheap body locate; a selector guard
needs a real DOM probe, whose timeout is kept short — a guard element that
exists shows up immediately, so most of the timeout is only ever spent
confirming absence. A guard defining both trips when either is present.
"""
signed_out = step.get("signedOut", "") or ""
pattern = step.get("urlPattern", "") or opts.url
open_url = step.get("openUrl", "") or ""
while time.time() < deadline:
if run_status(dash, run_id) == "cancelled":
return False
time.sleep(3.0)
if guard.get("urlContains"):
try:
landed = dash.locate("body", 0, pattern, opts.timeout, open_url)
except DashboardError:
continue # tab busy mid-redirect; look again shortly
if signed_out not in (landed.get("url") or ""):
return None # tab busy mid-redirect; the caller decides
if guard["urlContains"] in (landed.get("url") or ""):
return True
if not guard.get("selector"):
return False
if guard.get("selector"):
try:
dash.locate(guard["selector"], 0, pattern, probe_timeout, open_url)
return True
except NotFoundError:
return False
except DashboardError:
return None
return False
def signed_in_now(dash: Dashboard, step: dict, opts, settle_seconds: float = 20.0) -> bool:
"""Did we actually get back in?
def guard_cleared(dash: Dashboard, guard: dict, step: dict, opts,
settle_seconds: float = 20.0) -> bool:
"""Did recovery actually clear the condition?
Auth steps can 'succeed' — every click landing — while the site still rejects
the sign-in, so this confirms rather than assumes before the failed step is
retried.
Recovery steps can 'succeed' — every click landing — while the page stays in
the guarded state, so this confirms rather than assumes before resuming.
It polls rather than checking once: submitting a login form starts a network
round trip and then a redirect, so the tab is still sitting on the login URL
for a second or two afterwards. Checking immediately reports a failure for a
sign-in that is merely still in flight.
round trip and then a redirect, so the tab is still sitting on the guard
page for a second or two afterwards. Checking immediately reports a failure
for a recovery that is merely still in flight.
"""
signed_out = step.get("signedOut", "") or ""
if not signed_out:
return True
deadline = time.time() + settle_seconds
while True:
try:
landed = dash.locate("body", 0, step.get("urlPattern", "") or opts.url,
opts.timeout, step.get("openUrl", "") or "")
if signed_out not in (landed.get("url") or ""):
return True
except DashboardError:
pass # mid-redirect the tab can be briefly un-injectable
if guard_present(dash, guard, step, opts) is False:
return True
if time.time() >= deadline:
return False
time.sleep(1.0)
def wait_for_clear(dash: Dashboard, guard: dict, step: dict, run_id: int, opts,
deadline: float) -> bool:
"""Block until a human deals with the guard's condition.
The runner does not act here — for the login guard, entering credentials is
the human's job, and a keystroke aimed at the wrong field is not something
to risk automating. All this does is watch the page and pick the run back
up once the condition is gone.
"""
while time.time() < deadline:
if run_status(dash, run_id) == "cancelled":
return False
time.sleep(3.0)
if guard_present(dash, guard, step, opts) is False:
return True
return False
def find_tripped_guard(dash: Dashboard, guards: list, step: dict, opts) -> dict | None:
"""After a step failure: is a guard's condition what actually broke it?
URL guards are caught in flight by check_guards; selector guards cost a
locate round trip each, so they are only probed once a step has already
failed — a session-expired modal explains a NotFound far better than a
wrong selector does.
"""
for guard in guards:
if not guard.get("selector"):
continue # URL-only guards were already checked in flight
if guard_present(dash, guard, step, opts):
return guard
return None
def legacy_guards(run: dict) -> list:
"""Guards synthesized from the pre-guard payload shape, so this runner still
works against a dashboard that predates run-level guards."""
for step in run.get("steps") or []:
if step.get("signedOut"):
return [{
"id": "signed-out",
"label": "signed out",
"urlContains": step["signedOut"],
"resume": "continue",
"waitSeconds": float(step.get("signedOutWait", 0) or 0),
"recoverySteps": run.get("authSteps") or [],
}]
return []
def handle_guard(dash: Dashboard, run_id: int, step_index: int, label: str,
guard: dict, step: dict, opts, rng, report,
trigger_counts: dict) -> str:
"""A guard tripped mid-run: recover, then say where the run picks up.
Returns "retry" (redo the step that tripped), "restart" (back to step 1),
"cancelled", or "fail" — the terminal outcomes have already been reported
and the run finished by the time they are returned.
"""
gid = guard.get("id") or "guard"
name = guard.get("label") or gid
def fail(msg: str) -> str:
post_progress(dash, run_id, step_index, label, False, msg)
finish(dash, run_id, msg)
print(f"✗ run #{run_id} stopped: {msg}", file=sys.stderr)
return "fail"
# A guard whose recovery "works" without clearing the cause would trip
# again on the next step, forever. Cap it per run, not per step: a restart
# revisits every step, so a per-step cap would never fill up.
trigger_counts[gid] = trigger_counts.get(gid, 0) + 1
limit = int(guard.get("maxTriggers") or 3)
if trigger_counts[gid] > limit:
return fail(f"{name} guard tripped {trigger_counts[gid]} times in one run — giving up")
recovery = guard.get("recoverySteps") or []
if recovery:
post_progress(dash, run_id, step_index,
f"{name} guard tripped — running {len(recovery)} recovery step(s)",
False, None)
try:
for rstep in recovery:
rlabel = rstep.get("label") or f"recovery: {rstep['action']}"
print(f" [{name}] {rlabel}")
# No guards passed: recovery runs on the guard page itself.
run_one_step(dash, rstep, opts, rng, report, run_id)
except (actions.StepError, DashboardError) as exc:
return fail(f"{name} guard recovery failed: {exc}")
if not guard_cleared(dash, guard, step, opts):
return fail(f"{name} guard recovery ran but the condition is still on the page")
report(f"{name} guard cleared")
else:
wait_seconds = float(guard.get("waitSeconds") or 0)
if wait_seconds <= 0:
return fail(f"{name} guard tripped and defines no recovery — stopping")
mins = wait_seconds / 60
note = f"{name} guard tripped — waiting up to {mins:.0f} min for you to deal with it"
print(f" {note}", file=sys.stderr)
post_progress(dash, run_id, step_index, note, False, None)
if not wait_for_clear(dash, guard, step, run_id, opts, time.time() + wait_seconds):
if run_status(dash, run_id) == "cancelled":
print(f"■ run #{run_id} cancelled while waiting on the {name} guard")
return "cancelled"
return fail(f"{name} guard still tripped after {mins:.0f} min")
report(f"{name} guard cleared")
return "restart" if (guard.get("resume") or "continue") == "restart" else "retry"
def run_steps(dash: Dashboard, run: dict, opts) -> None:
"""Work through one automation. Reports every step; stops on the first
failure, because a half-completed purchase flow should not barrel on."""
failure, because a half-completed purchase flow should not barrel on.
Page guards can interrupt any step. Once one is handled, its `resume`
policy decides where the run picks up: "continue" retries the step that
tripped, "restart" goes back to the first step — hence a while loop over
the step index rather than a plain for."""
run_id = run["id"]
steps = run["steps"]
auth_steps = run.get("authSteps") or []
guards = run.get("guards")
if guards is None:
guards = legacy_guards(run)
rng = random.Random(opts.seed) if opts.seed is not None else random.Random()
print(f"\n▶ run #{run_id} — {run['label']} ({len(steps)} steps)")
for i, step in enumerate(steps):
trigger_counts: dict[str, int] = {}
i = 0
while i < len(steps):
step = steps[i]
label = step.get("label") or f"{step['action']} {step.get('selector', '')}".strip()
print(f" [{i + 1}/{len(steps)}] {label}")
@@ -300,74 +433,46 @@ def run_steps(dash: Dashboard, run: dict, opts) -> None:
detail_parts.append(message)
print(f" {message}")
# One retry after a sign-in: the session can only be expired once per step.
for attempt in range(2):
try:
run_one_step(dash, step, opts, rng, report, run_id)
break
outcome = None
try:
run_one_step(dash, step, opts, rng, report, run_id, guards=guards)
except actions.SignedOutError as exc:
wait_seconds = float(step.get("signedOutWait", 0) or 0)
except actions.GuardTripped as exc:
print(f" {exc}", file=sys.stderr)
outcome = handle_guard(dash, run_id, i, label, exc.guard, step,
opts, rng, report, trigger_counts)
# Scripted auth first, if the firm defines any; otherwise fall
# back to pausing for a human. Only ever attempted once per step —
# auth that "succeeds" without signing in would otherwise loop.
if auth_steps and attempt == 0:
post_progress(dash, run_id, i, f"signed out — running {len(auth_steps)} auth step(s)",
False, exc.args[0])
try:
for astep in auth_steps:
alabel = astep.get("label") or f"auth: {astep['action']}"
print(f" [auth] {alabel}")
run_one_step(dash, astep, opts, rng, report, run_id)
except (actions.StepError, DashboardError) as aexc:
msg = f"auth steps failed: {aexc}"
post_progress(dash, run_id, i, label, False, msg)
finish(dash, run_id, msg)
print(f"✗ run #{run_id} stopped: {msg}", file=sys.stderr)
return
except (actions.StepError, DashboardError) as exc:
if run_status(dash, run_id) == "cancelled":
print(f"■ run #{run_id} cancelled from the dashboard")
return
if not signed_in_now(dash, step, opts):
msg = "auth steps ran but the session is still signed out"
post_progress(dash, run_id, i, label, False, msg)
finish(dash, run_id, msg)
print(f"✗ run #{run_id} stopped: {msg}", file=sys.stderr)
return
report("signed in via auth steps — retrying")
continue
if wait_seconds <= 0 or attempt > 0:
print(f" FAILED: {exc}", file=sys.stderr)
post_progress(dash, run_id, i, label, False, str(exc))
finish(dash, run_id, str(exc))
print(f"✗ run #{run_id} stopped at step {i + 1}")
return
mins = wait_seconds / 60
note = f"signed out — waiting up to {mins:.0f} min for you to log in"
print(f" {note}", file=sys.stderr)
post_progress(dash, run_id, i, note, False, exc.args[0])
if not wait_for_sign_in(dash, step, run_id, opts, time.time() + wait_seconds):
if run_status(dash, run_id) == "cancelled":
print(f"■ run #{run_id} cancelled while waiting for sign-in")
return
msg = f"still signed out after {mins:.0f} min"
post_progress(dash, run_id, i, label, False, msg)
finish(dash, run_id, msg)
print(f"✗ run #{run_id} gave up waiting for sign-in")
return
report("signed in — retrying the step")
except (actions.StepError, DashboardError) as exc:
# Before writing the failure off, see whether a selector guard's
# condition is sitting on the page — an expired-session modal
# explains a missing button better than a wrong selector does.
tripped = find_tripped_guard(dash, guards, step, opts) if guards else None
if tripped is not None:
name = tripped.get("label") or tripped.get("id")
print(f" step failed with the {name} guard's condition on the page",
file=sys.stderr)
outcome = handle_guard(dash, run_id, i, label, tripped, step,
opts, rng, report, trigger_counts)
else:
print(f" FAILED: {exc}", file=sys.stderr)
post_progress(dash, run_id, i, label, False, str(exc))
finish(dash, run_id, str(exc))
print(f"✗ run #{run_id} stopped at step {i + 1}")
return
if outcome == "retry":
continue # same step, another go
if outcome == "restart":
print(f" ↺ restarting run #{run_id} from step 1")
i = 0
continue
if outcome in ("fail", "cancelled"):
return # handle_guard already reported and finished
status = post_progress(dash, run_id, i + 1, label, True, "; ".join(detail_parts) or None)
# The dashboard's Stop button shows up here.
@@ -375,6 +480,8 @@ def run_steps(dash: Dashboard, run: dict, opts) -> None:
print(f"■ run #{run_id} cancelled from the dashboard")
return
i += 1
finish(dash, run_id, None)
print(f"✓ run #{run_id} complete")