diff --git a/.agents/skills/sstack-demo b/.agents/skills/sstack-demo new file mode 120000 index 0000000..7e53a8b --- /dev/null +++ b/.agents/skills/sstack-demo @@ -0,0 +1 @@ +../../tools/sstack/skills/sstack-demo \ No newline at end of file diff --git a/.claude/skills/sstack-demo b/.claude/skills/sstack-demo new file mode 120000 index 0000000..7e53a8b --- /dev/null +++ b/.claude/skills/sstack-demo @@ -0,0 +1 @@ +../../tools/sstack/skills/sstack-demo \ No newline at end of file diff --git a/.github/workflows/sstack-quality.yml b/.github/workflows/sstack-quality.yml index 0cc2c54..18fa3d4 100644 --- a/.github/workflows/sstack-quality.yml +++ b/.github/workflows/sstack-quality.yml @@ -29,8 +29,10 @@ jobs: cache: pip cache-dependency-path: tools/sstack/requirements-dev.txt - run: python -m pip install -r tools/sstack/requirements-dev.txt + - run: python -m playwright install --with-deps chromium - name: Verify affected projects env: + SSTACK_REQUIRE_BROWSER: '1' BASE_REVISION: ${{ github.event.pull_request.base.sha || github.event.before }} run: | if [[ -z "$BASE_REVISION" || "$BASE_REVISION" =~ ^0+$ ]]; then diff --git a/AGENTS.md b/AGENTS.md index 1d63f01..f16ffd9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -8,6 +8,10 @@ Use /sstack plus a project goal for PM delivery. Read tools/sstack/skills/sstack Resolve the target project, its instructions and configured PM board before posting. Do not default to a previous customer, board, environment or credential. +Use /sstack-demo for demos. Demo work stays on demo/* branches, never merges to +the default branch and is never ticket completion evidence. It follows that +skill's own checks instead of delivery review and gates. + Use independent engineer/reviewer agents for meaningful changes. Parallel writers need isolated Git worktrees and nonconflicting live resources. Routine scoped validation, commits, pushes and PRs proceed within the authorized task. Respect diff --git a/CLAUDE.md b/CLAUDE.md index bd29249..bcae2a8 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -2,5 +2,6 @@ @AGENTS.md -Use /sstack for PM goals. Supporting skills: /sstack-engineer, /sstack-mode, -and /sstack-verify. Canonical files live in tools/sstack/skills. +Use /sstack for PM goals and /sstack-demo for demos. Supporting skills: +/sstack-engineer, /sstack-mode, and /sstack-verify. Canonical files live in +tools/sstack/skills. diff --git a/README.md b/README.md index 80e9d0d..692488b 100644 --- a/README.md +++ b/README.md @@ -168,6 +168,7 @@ their tools, credentials, conversation history and running agents remain separat | Skill | Purpose | | --- | --- | | `sstack` | PM intake, Jira breakdown, dispatch and delivery coordination | +| `sstack-demo` | Scripted, honestly labeled demos and their promotion to tickets | | `sstack-engineer` | Implement a scoped ticket and return tested PR evidence | | `sstack-mode` | Research, design, prototypes and direct project work | | `sstack-verify` | Validate behavior and maintain project verification recipes | diff --git a/install.py b/install.py index f300e26..88854d7 100644 --- a/install.py +++ b/install.py @@ -12,7 +12,7 @@ config = importlib.util.module_from_spec(CONFIG_SPEC) CONFIG_SPEC.loader.exec_module(config) -SKILLS = ("sstack", "sstack-engineer", "sstack-mode", "sstack-verify") +SKILLS = ("sstack", "sstack-demo", "sstack-engineer", "sstack-mode", "sstack-verify") START = "" END = "" IGNORE_BLOCK = """# sstack generated files @@ -25,8 +25,9 @@ """ BLOCK = f"""{START} Use /sstack for PM goals. Read tools/sstack/skills/sstack/SKILL.md and resolve -project scope and the configured PM board before dispatch. Supporting skills: -sstack-engineer, sstack-mode, sstack-verify. Canonical skills: tools/sstack/skills/. +project scope and the configured PM board before dispatch. Use /sstack-demo for +demos. Supporting skills: sstack-engineer, sstack-mode, sstack-verify. +Canonical skills: tools/sstack/skills/. Configure a board with python tools/sstack/control.py board set URL --project-key KEY. Run python tools/sstack/control.py board show before starting a new goal. {END} diff --git a/tools/sstack/README.md b/tools/sstack/README.md index e29a1e9..6d89009 100644 --- a/tools/sstack/README.md +++ b/tools/sstack/README.md @@ -64,6 +64,14 @@ time-dependent or shared-state checks. See the [delivery rules](skills/sstack/references/delivery.md#choose-the-shortest-sufficient-path) for required risk escalation and independent review. +## UI evaluation + +`ui_eval.py run` drives a JSON UI script in Chromium against one approved origin, +saving per-step screenshots, errors and captured values; `ui_eval.py review` binds +an agent's visual verdicts to those screenshots. See +[scripted UI evaluation](skills/sstack-verify/references/ui-eval.md). Playwright +is a development dependency; CI installs Chromium and requires the browser tests. + ## Project coverage and evidence `projects.json` registers owned paths, offline suites, external acceptance recipes diff --git a/tools/sstack/check_contracts.py b/tools/sstack/check_contracts.py index 8a2c54f..d9b6a5c 100644 --- a/tools/sstack/check_contracts.py +++ b/tools/sstack/check_contracts.py @@ -12,7 +12,7 @@ import yaml ROOT = Path(__file__).resolve().parents[2] -SKILLS = ("sstack", "sstack-engineer", "sstack-mode", "sstack-verify") +SKILLS = ("sstack", "sstack-demo", "sstack-engineer", "sstack-mode", "sstack-verify") STACK = Path("tools/sstack") FENCE = re.compile(r"^ {0,3}(`{3,}|~{3,})") # Inline destinations, including angle brackets, an optional title and one diff --git a/tools/sstack/requirements-dev.txt b/tools/sstack/requirements-dev.txt index 4bb163c..3c8bc46 100644 --- a/tools/sstack/requirements-dev.txt +++ b/tools/sstack/requirements-dev.txt @@ -1,2 +1,3 @@ ruff==0.12.12 PyYAML==6.0.2 +playwright==1.56.0 diff --git a/tools/sstack/skills/sstack-demo/SKILL.md b/tools/sstack/skills/sstack-demo/SKILL.md new file mode 100644 index 0000000..818be14 --- /dev/null +++ b/tools/sstack/skills/sstack-demo/SKILL.md @@ -0,0 +1,88 @@ +--- +name: sstack-demo +description: Build a presentable demo quickly from an agreed script, with scripted self-checks and an honest faked list. Use for demos and stakeholder previews, not for tickets, PRs, delivery, or technical prototypes. +--- + +# SStack demo + +A demo presents an agreed story to an audience reliably and honestly. It is not +delivery: skip Jira, the delivery controller, frozen eval plans, independent code +review, `verify changed`, CI and merge; this loop's checks replace them. Delivery +requests go to `sstack`; do not load its delivery or Jira references here. To +settle a technical uncertainty, use the `sstack-mode` Prototype workflow instead. +Root AGENTS.md secret, customer-data and approval rules still apply. + +## Boundaries + +- Work on a `demo/` branch with the script and assets under `demos//`. + The branch never merges and is never completion evidence for a ticket. +- Faked steps are labeled disposable prototypes under engineering rule 1 in + [engineering principles](../sstack-mode/references/engineering.md). +- Use synthetic data only. Read + [platform](../sstack-verify/references/platform.md) before live Kizen + interaction. Record writes go only to a sandbox the user approved for this demo; + any Kizen configuration change follows platform.md's dry-run and approval rules. + Without an approved sandbox, fake the step or ask for access. +- Never present a step as real without proof that it ran against the real system. + +## 1. Script before building + +Write `demos//script.md` with the audience, time budget, and 3–7 steps: + +```markdown +| # | Action | Visible result | Real or faked | Proof or fake detail | +| --- | --- | --- | --- | --- | +| 1 | Enter $6,000 stocks and $4,000 bonds | Chart shows 60% / 40% | Real | Sandbox record readback | +| 2 | Refresh prices | Prices update | Faked | Hardcoded price fixture | +``` + +Confirm the script with the requester when intent is unclear; otherwise start. +Script changes from stakeholder feedback are expected, not violations. + +## 2. Build and self-check + +Reuse existing app code, templates and seed data before writing new ones. Start from +a clean app state with a fresh run marker; leave earlier sandbox records in place +rather than deleting them. Encode the script as a +[UI script](../sstack-verify/references/ui-eval.md) and run `ui_eval.py run` +against the approved origin; add `--video` for a recording. It checks each step, +saves a screenshot per step, and fails on console errors or failed requests. + +Then do the visual check: view every screenshot, with computer use or by opening +the image, against the script's visible result and record verdicts with +`ui_eval.py review`. Fix and rerun until the review passes. + +## 3. Honesty check + +- Each real step has evidence it ran: for Kizen, read back the record it wrote, + matched by a marker unique to this run. A step without proof is relabeled faked. +- Every mock, stub or hardcoded response in the branch's diff from the default + branch appears in the faked column, and every faked entry exists in the code. +- No real customer data, credentials or unapproved writes. + +## 4. Rehearse + +Run `ui_eval.py run` twice more with fresh markers within the time budget. Both runs +must pass automated checks and visual review. For a customer-facing demo, optionally +ask a fresh agent to review only the recording and script: does it tell the story, +and what would confuse the audience? + +Report the branch, URL or launch command, recording or screenshots, rehearsal +results and the faked list. Stakeholder feedback returns to step 1. + +## 5. Promote or archive + +When a stakeholder approves, hand off to `sstack` to promote the demo into tickets: + +- Each real step becomes an acceptance ID with an eval case: its input, expected + business values (for example `{"stocks_pct": 60}`, not the displayed text), + phase `premerge` or `postmerge`, and surface `local`, `live` or `ui` for the + proof delivery will collect. +- Each faked step becomes scope with its own acceptance criteria and eval cases. +- Link the script and recording as the agreed user journey. They are references, + not result evidence; design still follows the `sstack` Kizen-first principles. +- The engineer builds to delivery standards and may borrow demo code, but must + replace every faked-list item with a real path. + +Archive a rejected demo by pushing its branch unchanged, so the script and +recording links stay valid. Nothing enters Jira. diff --git a/tools/sstack/skills/sstack-mode/SKILL.md b/tools/sstack/skills/sstack-mode/SKILL.md index 208d0d8..2a50c47 100644 --- a/tools/sstack/skills/sstack-mode/SKILL.md +++ b/tools/sstack/skills/sstack-mode/SKILL.md @@ -10,7 +10,8 @@ gate with `control.py verify changed --base --jobs 2`; avoid duplicating lint and tests that the selected suites already include. For a project goal/backlog, use `sstack`; for a dispatched issue use -`sstack-engineer`. Ordinary questions and small direct edits need no new hierarchy. +`sstack-engineer`; for a demo use `sstack-demo`. Ordinary questions and small +direct edits need no new hierarchy. Read skills at `tools/sstack/skills//SKILL.md` when not auto-discovered. 1. Restate the user outcome and observable acceptance. Read affected source and diff --git a/tools/sstack/skills/sstack-verify/SKILL.md b/tools/sstack/skills/sstack-verify/SKILL.md index e04efde..80546d2 100644 --- a/tools/sstack/skills/sstack-verify/SKILL.md +++ b/tools/sstack/skills/sstack-verify/SKILL.md @@ -37,7 +37,8 @@ around existing suites does not establish their completeness. Reproduce the actual user input/action. Assert observable values/writes/no-writes for pure logic. For platform work inspect current state, exact execution and child runs, and the corresponding persisted business result. For UI work drive the -documented user path with available browser tools and inspect rendered output. +documented user path with [scripted UI evaluation](references/ui-eval.md) or +available browser tools, and view every rendered step, not only its assertions. Capture the trigger and outcome, not just a final screen. Use synthetic records. Separate local, live Kizen, UI, and untested paths. A build is not live proof; diff --git a/tools/sstack/skills/sstack-verify/references/ui-eval.md b/tools/sstack/skills/sstack-verify/references/ui-eval.md new file mode 100644 index 0000000..b0140a5 --- /dev/null +++ b/tools/sstack/skills/sstack-verify/references/ui-eval.md @@ -0,0 +1,75 @@ +# Scripted UI evaluation + +`tools/sstack/ui_eval.py` drives a JSON UI script in Chromium, screenshots every +step, and records console errors, page errors, failed requests and captured +values in one `report.json`. An agent then views each screenshot and records a +visual verdict. It is built for web UIs such as the Kizen UI but has not yet been +exercised against a live Kizen login (SSO, MFA, session expiry). It does not call +the Kizen API or prove that a record was persisted. + +Install `tools/sstack/requirements-dev.txt` (includes Playwright) and, where no +Chromium is preinstalled, run `python -m playwright install chromium`. + +## Script + +```json +{"schema_version": 1, "base_url": "https://sandbox.example.test", + "viewport": {"width": 1440, "height": 900}, "ignore_console": ["^Known noise$"], + "steps": [ + {"id": "open", "action": "goto", "path": "/records"}, + {"action": "fill", "selector": "role=textbox[name=\"Name\"]", "value": "Demo {run_marker}"}, + {"id": "save", "action": "click", "selector": "role=button[name=\"Save\"]"}, + {"action": "expect_text", "selector": "role=main", "text": "Demo {run_marker}"}, + {"id": "chart", "action": "capture", "selector": "#allocation", "case": "mixed-holdings", + "key": "stocks_pct", "as": "number"} + ]} +``` + +Actions: `goto` (path relative to `base_url`), `click`, `fill`, `press`, +`wait_for`, `expect_text` and `capture`. Selectors use Playwright syntax; prefer +role and visible-text selectors over generated class names. `{run_marker}` is +replaced by a value unique to each run, so readback can match this run's records. +The first failed step stops the run; later steps are reported as skipped. +`capture` with `"as": "number"` needs exactly one standalone number in the text; +ranges, dates and numbers with attached units such as `5px` are rejected. +`ignore_console` patterns only suppress known console noise; they never hide page +errors or failed requests. Failed requests are network failures and 5xx responses +on the approved origin, plus 4xx responses for its top-level pages. + +## Run and review + +```sh +python tools/sstack/ui_eval.py run SCRIPT.json --approved-origin https://sandbox.example.test \ + --storage-state /outside/repo/ui-state.json [--video] [--headed] +python tools/sstack/ui_eval.py review artifacts/sstack/ui//report.json VERDICTS.json +``` + +- `--approved-origin` is the human-approved UI host. The script's `base_url` must + match it and `goto` paths must stay under it. Top-level navigation elsewhere, + including links, server redirects at any hop, delayed scripts and popups, is + aborted before the request is sent and fails the run. The tool fetches top-level + pages itself and turns each same-origin redirect into a new checked navigation; + chains over 20 hops and 307/308 redirects of form posts fail. The run waits + briefly after the last step to catch late navigation. A login redirect to another + host fails the run, so start from a saved session. UI actions can write data, so + the origin needs the same approval as any other write there; for Kizen follow + [platform](platform.md) and use the confirmed UI host, which may differ from the + API host. +- `--storage-state` is a saved login session and must live outside the repository. + Create one with `npx playwright codegen --save-storage= ` and log + in as the intended role. Never commit or share it. +- Visual review: view every `step-NN-.png` yourself, with computer use or by + opening the image, and compare it to the step's expected visible result. Check + layout, labels, values and error states that the automated checks cannot see. + Write `{"reviewer": "", "steps": {"": {"verdict": "pass", "note": + "..."}}}` covering every screenshotted step, then run `review`. It fails on any + `fail` verdict or failed automated check, blocks on missing verdicts, and refuses + screenshots changed since the run. Verdicts are the reviewer's declarations; the + tool binds them to the screenshot hashes but cannot authenticate them. + +`report.json` `eval_cases` already has the result-case shape (`id`, `observed`, +`evidence`) for `surface: "ui"` cases in the +[evaluation contract](../../sstack/references/jira-handoff.md#pm-owned-evaluation-sets). +Rendered values are UI proof only; persisted Kizen results still need readback. +Recorded URLs drop query strings and fragments. Reports and screenshots can still +show record data; use synthetic records and inspect them before sharing. diff --git a/tools/sstack/skills/sstack/SKILL.md b/tools/sstack/skills/sstack/SKILL.md index 137867f..8cc2023 100644 --- a/tools/sstack/skills/sstack/SKILL.md +++ b/tools/sstack/skills/sstack/SKILL.md @@ -5,6 +5,9 @@ description: Break requested project goals into Jira work, design with Kizen fir # SStack PM +Demo requests go to `sstack-demo` instead; promote an approved demo using its +step 5. + Read [Jira handoff](references/jira-handoff.md) and [delivery gates](references/delivery.md). Follow root AGENTS.md for shared preferences and host switching. Use delivery.md's shortest sufficient path: one diff --git a/tools/sstack/test_install.py b/tools/sstack/test_install.py index ebc270e..64ed0f3 100644 --- a/tools/sstack/test_install.py +++ b/tools/sstack/test_install.py @@ -35,7 +35,7 @@ def install(self, **kwargs): def test_install_preserves_instructions_and_rerun_is_noop(self): result = self.install(board_url="https://boards.test/projects/PM", project_key="PM") - self.assertEqual(len(result["links"]), 8) + self.assertEqual(len(result["links"]), 2 * len(installer.SKILLS)) self.assertTrue((self.target / ".claude/skills/sstack/SKILL.md").is_file()) self.assertTrue((self.target / "AGENTS.md").read_text().startswith("# User instructions\n")) self.assertFalse((self.target / "tools/sstack/untracked-secret.txt").exists()) diff --git a/tools/sstack/test_ui_eval.py b/tools/sstack/test_ui_eval.py new file mode 100644 index 0000000..25b3bb0 --- /dev/null +++ b/tools/sstack/test_ui_eval.py @@ -0,0 +1,301 @@ +import json +import os +import tempfile +import threading +import unittest +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +import ui_eval + +PAGES = { + "/": """Allocation + + +

+Away + + + + +""", +} + + +class Handler(BaseHTTPRequestHandler): + def do_GET(self): + self.server.hosts.append(self.headers["Host"].split(":")[0]) + path = self.path.split("?")[0] + redirects = { + "/home": "/", + "/chain": "/home", + "/leave": f"http://localhost:{self.server.server_port}/x", + "/hop": "/leave", + "/loop": "/loop", + } + if path in redirects: + self.send_response(302) + self.send_header("Location", redirects[path]) + self.end_headers() + return + page = PAGES.get(path) + status = 200 if page else 500 if path == "/boom" else 404 + body = (page or "missing").replace("{port}", str(self.server.server_port)).encode() + self.send_response(status) + self.send_header("Content-Type", "text/html") + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args): + pass + + +def script(base, *steps): + return { + "schema_version": 1, + "base_url": base, + "steps": [{"action": "goto", "path": "/"}, *steps], + } + + +def capture(**extra): + return {"action": "capture", "selector": "x", "case": "c", "key": "k", **extra} + + +class ValidationTests(unittest.TestCase): + def test_rejects_unsafe_or_malformed_scripts(self): + base = "https://a.test" + cases = [ + {"schema_version": 1, "base_url": "http://example.com", "steps": [{}]}, + *( + script(base, {"action": "goto", "path": path}) + for path in ( + "https://b.test/", + "//b.test/", + "/\\b.test/", + "http:b.test/x", + "file:///etc/passwd", + "javascript:alert(1)", + "records", + ) + ), + script(base, {"action": "click"}), + script(base, {"action": "eval", "selector": "x"}), + script(base, {"action": "click", "selector": "x", "id": "bad id"}), + script(base, capture(**{"as": "html"})), + script(base, capture(), capture(id="again")), + script(base, capture(case=" c")), + {**script(base), "ignore_console": ["("]}, + {**script(base), "extra": True}, + ] + for case in cases: + with self.subTest(case=case), self.assertRaises(ValueError): + ui_eval.validate_script(case) + ui_eval.validate_script(script("http://127.0.0.1:8000", capture())) + + def test_storage_state_must_live_outside_repository(self): + with tempfile.TemporaryDirectory() as root: + inside = Path(root, "state.json") + inside.write_text("{}") + with self.assertRaises(ValueError): + ui_eval.check_storage_state(inside, root) + with tempfile.NamedTemporaryFile(suffix=".json") as outside: + self.assertEqual( + ui_eval.check_storage_state(outside.name, root), Path(outside.name).resolve() + ) + + def test_numeric_capture_takes_exactly_one_standalone_number(self): + for text, expected in [ + ("Stocks 60%", 60), + ("60.0%", 60), + ("$1,234.50", 1234.5), + ("Q3 total 60", 60), + ("Item-5 total 60", 60), + ("Change -12", -12), + ("Change −12", -12), + ("(1,234)", -1234), + ]: + with self.subTest(text=text): + self.assertEqual(ui_eval.captured(text, "number"), expected) + for text in ("none", "1,2,3", "1.234,5", "60% of 100", "5-10", "2024-01-05", "1e5"): + with self.subTest(text=text), self.assertRaises(ValueError): + ui_eval.captured(text, "number") + + def test_review_rejects_reports_without_reviewable_screenshots(self): + with tempfile.TemporaryDirectory() as folder: + report = Path(folder, "report.json") + verdicts = {"reviewer": "r", "steps": {}} + for content in ( + [], + {"status": "passed"}, + {"status": "passed", "steps": []}, + {"status": "passed", "steps": [{"status": "passed"}]}, + {"status": "passed", "steps": [{"id": "a", "status": "passed"}]}, + ): + report.write_text(json.dumps(content)) + with self.subTest(content=content), self.assertRaises(ValueError): + ui_eval.review(report, verdicts) + + +class BrowserTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + try: + from playwright.sync_api import Error, sync_playwright + + with sync_playwright() as playwright: + playwright.chromium.launch().close() + except (ImportError, Error) as error: + if os.environ.get("SSTACK_REQUIRE_BROWSER"): + raise + raise unittest.SkipTest(f"Chromium unavailable: {type(error).__name__}") from error + cls.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + cls.server.hosts = [] + threading.Thread(target=cls.server.serve_forever, daemon=True).start() + cls.base = f"http://127.0.0.1:{cls.server.server_port}" + + @classmethod + def tearDownClass(cls): + cls.server.shutdown() + cls.server.server_close() + + def setUp(self): + temp = tempfile.TemporaryDirectory() + self.addCleanup(temp.cleanup) + self.out = Path(temp.name) + self.server.hosts.clear() + + def run_script(self, *steps, **extra): + return ui_eval.run( + {**script(self.base, *steps), **extra}, approved_origin=self.base, out_dir=self.out + ) + + def click(self, selector, **extra): + return {"action": "click", "selector": selector, **extra} + + def test_scripted_path_captures_values_screenshots_and_review(self): + report = self.run_script( + {"action": "fill", "selector": "#stocks", "value": "6000"}, + {"action": "fill", "selector": "#bonds", "value": "4000"}, + {"action": "fill", "selector": "#note", "value": "{run_marker}"}, + self.click("#go", id="show"), + {"action": "expect_text", "selector": "#owner", "text": "{run_marker}"}, + { + "action": "capture", + "selector": "#result", + "case": "mixed", + "key": "stocks_pct", + "as": "number", + "id": "chart", + }, + ) + self.assertEqual(report["status"], "passed", report) + self.assertEqual(report["visual_review"], "pending") + self.assertEqual( + report["eval_cases"], + [ + { + "id": "mixed", + "observed": {"stocks_pct": 60}, + "evidence": ["ui-screenshot:step-07-chart.png"], + } + ], + ) + self.assertTrue(all((self.out / step["screenshot"]).is_file() for step in report["steps"])) + saved = json.loads((self.out / "report.json").read_text()) + self.assertEqual(saved["run_marker"], report["run_marker"]) + + ids = [step["id"] for step in report["steps"]] + verdict = {"verdict": "pass", "note": "Chart label reads 60%"} + partial = {"reviewer": "demo-verifier", "steps": {"chart": verdict}} + self.assertEqual(ui_eval.review(self.out / "report.json", partial)["status"], "blocked") + complete = {"reviewer": "demo-verifier", "steps": dict.fromkeys(ids, verdict)} + self.assertEqual(ui_eval.review(self.out / "report.json", complete)["status"], "passed") + complete["steps"]["chart"] = {"verdict": "fail", "note": "Label clipped"} + rejected = ui_eval.review(self.out / "report.json", complete) + self.assertEqual(rejected["reasons"], ["visual_fail:chart"]) + (self.out / "step-07-chart.png").write_bytes(b"swapped") + with self.assertRaises(ValueError): + ui_eval.review(self.out / "report.json", complete) + + def test_failed_expectation_stops_later_steps_and_fails_review(self): + report = self.run_script( + {"action": "expect_text", "selector": "#result", "text": "never", "timeout_ms": 300}, + self.click("#go"), + ) + self.assertEqual(report["status"], "failed") + self.assertEqual([s["status"] for s in report["steps"]], ["passed", "failed", "skipped"]) + self.assertIn("screenshot", report["steps"][1]) + ids = [step["id"] for step in report["steps"][:2]] + verdicts = {"reviewer": "r", "steps": dict.fromkeys(ids, {"verdict": "pass", "note": "ok"})} + result = ui_eval.review(self.out / "report.json", verdicts) + self.assertEqual( + (result["status"], result["reasons"]), ("failed", ["automated_checks_failed"]) + ) + + def test_console_errors_fail_unless_ignored(self): + noisy = self.click("#noisy") + self.assertEqual(self.run_script(noisy)["console_errors"], ["boom"]) + report = self.run_script(noisy, ignore_console=["^boom$"]) + self.assertEqual(report["status"], "passed", report) + + def test_page_errors_and_server_errors_fail_with_redacted_urls(self): + wait = {"action": "wait_for", "selector": "#go", "timeout_ms": 300} + thrown = self.run_script(self.click("#throw"), wait, wait, ignore_console=[".*"]) + self.assertEqual(thrown["status"], "failed") + self.assertIn("kaput", thrown["page_errors"][0]) + broken = self.run_script(self.click("#broken"), wait, wait, ignore_console=[".*"]) + self.assertEqual(broken["status"], "failed") + self.assertEqual(broken["failed_requests"], [f"500 {self.base}/boom?…"]) + self.assertNotIn("secret", json.dumps(broken)) + + def test_missing_top_level_page_fails(self): + report = ui_eval.run( + { + "schema_version": 1, + "base_url": self.base, + "steps": [{"action": "goto", "path": "/x"}], + }, + approved_origin=self.base, + out_dir=self.out, + ) + self.assertEqual(report["status"], "failed") + self.assertEqual(report["failed_requests"], [f"404 {self.base}/x"]) + + def test_off_origin_navigation_is_blocked_before_any_request(self): + for trigger in ("#away", "#later", "#popup"): + with self.subTest(trigger=trigger): + self.server.hosts.clear() + wait = {"action": "wait_for", "selector": "#go", "timeout_ms": 300} + report = self.run_script(self.click(trigger), wait, wait) + self.assertEqual(report["status"], "failed", report) + self.assertTrue(report["blocked_navigations"], report) + self.assertNotIn("localhost", self.server.hosts) + + def test_redirects_are_checked_before_the_browser_follows_them(self): + chain = self.run_script({"action": "goto", "path": "/chain"}, self.click("#go")) + self.assertEqual(chain["status"], "passed", chain) + away = f"http://localhost:{self.server.server_port}/x" + for path in ("/leave", "/hop", "/loop"): + with self.subTest(path=path): + self.server.hosts.clear() + report = self.run_script({"action": "goto", "path": path}) + self.assertEqual(report["status"], "failed", report) + self.assertTrue(report["blocked_navigations"], report) + self.assertNotIn("localhost", self.server.hosts) + if path != "/loop": + self.assertEqual(report["blocked_navigations"], [away]) + + def test_navigation_after_the_last_step_is_caught(self): + report = self.run_script(self.click("#later")) + self.assertEqual(report["status"], "failed", report) + self.assertTrue(report["blocked_navigations"], report) + self.assertNotIn("localhost", self.server.hosts) + + def test_unapproved_origin_is_refused_before_launch(self): + with self.assertRaises(ValueError): + ui_eval.run(script(self.base), approved_origin="https://other.test", out_dir=self.out) diff --git a/tools/sstack/ui_eval.py b/tools/sstack/ui_eval.py new file mode 100644 index 0000000..e370895 --- /dev/null +++ b/tools/sstack/ui_eval.py @@ -0,0 +1,459 @@ +#!/usr/bin/env python3 +"""Drive a scripted browser path and record per-step UI evidence for visual review.""" + +from __future__ import annotations + +import argparse +import datetime +import hashlib +import json +import re +import secrets +import sys +import time +from pathlib import Path +from urllib.parse import urljoin, urlsplit + +ROOT = Path(__file__).resolve().parents[2] +ACTIONS = { + "goto": {"path"}, + "click": {"selector"}, + "fill": {"selector", "value"}, + "press": {"selector", "key"}, + "wait_for": {"selector"}, + "expect_text": {"selector", "text"}, + "capture": {"selector", "case", "key"}, +} +STEP_OPTIONAL = {"id", "timeout_ms"} +SCRIPT_REQUIRED = {"schema_version", "base_url", "steps"} +SCRIPT_OPTIONAL = {"viewport", "ignore_console"} +LOCAL_HOSTS = {"localhost", "127.0.0.1", "::1"} +MARKER = "{run_marker}" +# One standalone number: not glued to letters, other digits or separators. +NUMBER = re.compile(r"(?]*") +VERDICTS = {"pass", "fail"} +PROBLEMS = ("console_errors", "page_errors", "failed_requests", "blocked_navigations") + + +def origin(url): + parts = urlsplit(url) + return f"{parts.scheme}://{parts.netloc}".lower() + + +def redact(text): + """Drop URL queries and fragments, which can carry tokens or signed parameters.""" + return QUERY.sub("?…", text)[:300] + + +def _text(value, name): + if not isinstance(value, str) or not value.strip() or value != value.strip(): + raise ValueError(f"{name}: nonempty trimmed text required") + + +def validate_script(script): + if not isinstance(script, dict) or not SCRIPT_REQUIRED <= set(script): + raise ValueError("script: schema_version, base_url and steps required") + if set(script) - SCRIPT_REQUIRED - SCRIPT_OPTIONAL: + raise ValueError("script: unknown fields") + if script["schema_version"] != 1 or type(script["schema_version"]) is not int: + raise ValueError("script: unsupported schema_version") + parts = urlsplit(script["base_url"]) if isinstance(script["base_url"], str) else None + if ( + not parts + or not parts.netloc + or ( + parts.scheme != "https" + and not (parts.scheme == "http" and parts.hostname in LOCAL_HOSTS) + ) + ): + raise ValueError("script: base_url must be https, or http on localhost") + viewport = script.get("viewport", {"width": 1440, "height": 900}) + if not isinstance(viewport, dict) or set(viewport) != {"width", "height"}: + raise ValueError("script: viewport needs width and height") + if not all(type(value) is int and value > 0 for value in viewport.values()): + raise ValueError("script: viewport values must be positive integers") + ignore = script.get("ignore_console", []) + if not isinstance(ignore, list): + raise ValueError("script: ignore_console must be a list of patterns") + for pattern in ignore: + _text(pattern, "ignore_console") + try: + re.compile(pattern) + except re.error as error: + raise ValueError(f"ignore_console: invalid pattern {pattern!r}") from error + steps = script["steps"] + if not isinstance(steps, list) or not steps: + raise ValueError("script: steps required") + identifiers, captures = set(), set() + for index, step in enumerate(steps, 1): + if not isinstance(step, dict) or step.get("action") not in ACTIONS: + raise ValueError(f"step {index}: unknown action") + required = ACTIONS[step["action"]] + optional = STEP_OPTIONAL | ({"as"} if step["action"] == "capture" else set()) + if not required <= set(step) or set(step) - required - optional - {"action"}: + raise ValueError(f"step {index}: fields for {step['action']} are {sorted(required)}") + for key in required: + _text(step[key], f"step {index}.{key}") + if step["action"] == "goto" and ( + not step["path"].startswith("/") + or step["path"].startswith("//") + or "\\" in step["path"] + or origin(urljoin(script["base_url"], step["path"])) != origin(script["base_url"]) + ): + raise ValueError(f"step {index}: goto path must start with / under base_url") + if step.get("as", "text") not in {"text", "number"}: + raise ValueError(f"step {index}: capture as must be text or number") + if step["action"] == "capture": + if (step["case"], step["key"]) in captures: + raise ValueError(f"step {index}: duplicate capture of {step['case']}.{step['key']}") + captures.add((step["case"], step["key"])) + timeout = step.get("timeout_ms", 1) + if type(timeout) is not int or timeout <= 0: + raise ValueError(f"step {index}: timeout_ms must be a positive integer") + identifier = step.get("id", f"step-{index}") + _text(identifier, f"step {index}.id") + if not re.fullmatch(r"[A-Za-z0-9_-]+", identifier) or identifier in identifiers: + raise ValueError(f"step {index}: id must be unique letters, digits, - or _") + identifiers.add(identifier) + return script + + +def check_storage_state(path, root=ROOT): + resolved = Path(path).resolve() + if resolved == Path(root).resolve() or Path(root).resolve() in resolved.parents: + raise ValueError("storage state holds a login session; keep it outside the repository") + if not resolved.is_file(): + raise ValueError("storage state file not found") + return resolved + + +def captured(text, kind): + if kind == "text": + return text.strip() + normalized = text.replace("−", "-") + matches = list(NUMBER.finditer(normalized)) + if len(matches) != 1: + raise ValueError(f"capture_needs_one_number: {text.strip()[:80]!r}") + match = matches[0] + number = float(match.group().replace(",", "")) + before, after = normalized[: match.start()].rstrip(), normalized[match.end() :].lstrip() + if before.endswith("(") and after.startswith(")"): + number = -number + return int(number) if number.is_integer() else number + + +def digest(path): + return hashlib.sha256(Path(path).read_bytes()).hexdigest() + + +def run( + script, + *, + approved_origin, + out_dir, + storage_state=None, + headed=False, + video=False, + timeout_ms=15000, + root=ROOT, +): + """Run the script against the approved origin only; UI actions may write data.""" + from playwright.sync_api import Error as PlaywrightError + from playwright.sync_api import expect, sync_playwright + + validate_script(script) + approved = origin(approved_origin) + if origin(script["base_url"]) != approved: + raise ValueError("base_url origin differs from the approved origin") + if storage_state: + storage_state = check_storage_state(storage_state, root) + out = Path(out_dir) + out.mkdir(parents=True, exist_ok=True) + marker = "sstack-" + secrets.token_hex(4) + ignore = [re.compile(pattern) for pattern in script.get("ignore_console", [])] + problems = {kind: [] for kind in PROBLEMS} + hops = {"count": 0} + popups = [] + + def console(message): + if message.type == "error" and not any(p.search(message.text) for p in ignore): + problems["console_errors"].append(redact(message.text)) + + def top_level(request): + if not request.is_navigation_request(): + return False + try: + return request.frame.parent_frame is None + except PlaywrightError: + return True # A popup's first navigation precedes its frame. + + def response(item): + top = top_level(item.request) + if origin(item.url) == approved and (item.status >= 500 or (top and item.status >= 400)): + problems["failed_requests"].append(redact(f"{item.status} {item.url}")) + + def failed_request(request): + if origin(request.url) == approved: + problems["failed_requests"].append( + redact(f"{request.method} {request.url} {request.failure}") + ) + + def block(url, route): + problems["blocked_navigations"].append(redact(url)) + route.abort() + + def guard(route): + request = route.request + if not top_level(request): + route.continue_() + elif origin(request.url) != approved: + block(request.url, route) + else: + # Browsers follow network redirects without routing them. Fetch top-level pages + # here and replace each redirect with a new navigation, which is routed again. + try: + fetched = route.fetch(max_redirects=0) + except PlaywrightError as error: + problems["failed_requests"].append(redact(str(error).splitlines()[0])) + route.abort() + return + location = fetched.headers.get("location") + if not (300 <= fetched.status < 400 and location): + hops["count"] = 0 + route.fulfill(response=fetched) + return + target = urljoin(request.url, location) + hops["count"] += 1 + if origin(target) != approved: + block(target, route) + elif hops["count"] > 20 or (fetched.status in {307, 308} and request.method != "GET"): + block(f"unsupported redirect {fetched.status} to {target}", route) + else: + script = json.dumps(target).replace("<", "\\u003c") + route.fulfill( + content_type="text/html", + body=f"", + ) + + started_at = datetime.datetime.now(datetime.timezone.utc).isoformat() + started = time.monotonic() + steps, observed, evidence = [], {}, {} + with sync_playwright() as playwright: + browser = playwright.chromium.launch(headless=not headed) + options = {"viewport": script.get("viewport", {"width": 1440, "height": 900})} + if storage_state: + options["storage_state"] = str(storage_state) + if video: + options["record_video_dir"] = str(out / "video") + context = browser.new_context(**options) + context.route("**/*", guard) + context.on("console", console) + context.on( + "weberror", lambda error: problems["page_errors"].append(redact(str(error.error))) + ) + context.on("response", response) + context.on("requestfailed", failed_request) + page = context.new_page() + context.on("page", lambda opened: popups.append(redact(opened.url))) + failed = False + for index, step in enumerate(script["steps"], 1): + identifier = step.get("id", f"step-{index}") + record = {"index": index, "id": identifier, "action": step["action"]} + if failed: + steps.append({**record, "status": "skipped"}) + continue + timeout = step.get("timeout_ms", timeout_ms) + blocked = len(problems["blocked_navigations"]) + step_started = time.monotonic() + value = None + try: + selector = step.get("selector") + locator = page.locator(selector) if selector else None + action = step["action"] + if action == "goto": + path = step["path"].replace(MARKER, marker) + target = urljoin(script["base_url"], path) + if origin(target) != approved: + raise ValueError("off_origin_navigation: goto target") + page.goto(target, timeout=timeout) + elif action == "click": + locator.click(timeout=timeout) + elif action == "fill": + locator.fill(step["value"].replace(MARKER, marker), timeout=timeout) + elif action == "press": + locator.press(step["key"], timeout=timeout) + elif action == "wait_for": + locator.wait_for(timeout=timeout) + elif action == "expect_text": + expected = step["text"].replace(MARKER, marker) + expect(locator).to_contain_text(expected, timeout=timeout) + else: + value = captured(locator.inner_text(timeout=timeout), step.get("as", "text")) + if len(problems["blocked_navigations"]) > blocked: + raise ValueError("off_origin_navigation: blocked") + if origin(page.url) != approved: + raise ValueError(f"off_origin_navigation: {origin(page.url)}") + record["status"] = "passed" + except (PlaywrightError, AssertionError, ValueError) as error: + record["status"] = "failed" + record["error"] = redact(str(error).splitlines()[0]) + if origin(page.url) == approved: + screenshot = out / f"step-{index:02d}-{identifier}.png" + try: + page.screenshot(path=str(screenshot), full_page=True) + record["screenshot"] = screenshot.name + record["screenshot_sha256"] = digest(screenshot) + except PlaywrightError as error: + record["status"] = "failed" + record["screenshot_error"] = redact(str(error).splitlines()[0]) + else: + record["status"] = "failed" + record["screenshot_error"] = "page is off the approved origin" + if record["status"] == "passed" and step["action"] == "capture": + observed.setdefault(step["case"], {})[step["key"]] = value + record["observed"] = value + evidence.setdefault(step["case"], []).append(f"ui-screenshot:{screenshot.name}") + failed = record["status"] == "failed" + record["duration_seconds"] = round(time.monotonic() - step_started, 3) + steps.append(record) + if not failed: + try: + page.wait_for_load_state(timeout=timeout_ms) + page.wait_for_timeout(500) + except PlaywrightError as error: + problems["page_errors"].append(redact(str(error).splitlines()[0])) + browser_version = browser.version + context.close() + browser.close() + passed = all(step["status"] == "passed" for step in steps) and not any(problems.values()) + report = { + "schema_version": 1, + "status": "passed" if passed else "failed", + "visual_review": "pending", + "base_url": script["base_url"], + "script_sha256": hashlib.sha256( + json.dumps(script, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest(), + "run_marker": marker, + "browser": f"chromium {browser_version}", + "started_at": started_at, + "elapsed_seconds": round(time.monotonic() - started, 3), + "steps": steps, + **problems, + "popups": popups, + "eval_cases": [ + {"id": case, "observed": values, "evidence": evidence[case]} + for case, values in observed.items() + ], + "scope": "automated UI checks only; screenshots need visual review", + } + (out / "report.json").write_text(json.dumps(report, indent=2) + "\n") + return report + + +def _report_steps(report): + if not isinstance(report, dict) or report.get("status") not in {"passed", "failed"}: + raise ValueError("report: not a ui_eval run report") + steps = report.get("steps") + if not isinstance(steps, list) or not steps: + raise ValueError("report: steps required") + for step in steps: + if not isinstance(step, dict) or not isinstance(step.get("id"), str): + raise ValueError("report: malformed step") + if step.get("status") != "skipped" and not isinstance(step.get("screenshot"), str): + raise ValueError(f"report: step {step['id']} has no screenshot to review") + return {step["id"]: step for step in steps if step.get("status") != "skipped"} + + +def review(report_path, verdicts): + """Bind an agent's per-screenshot visual verdicts to the recorded evidence.""" + report_path = Path(report_path) + report = json.loads(report_path.read_text()) + shots = _report_steps(report) + if not isinstance(verdicts, dict) or set(verdicts) != {"reviewer", "steps"}: + raise ValueError("verdicts: reviewer and steps required") + _text(verdicts["reviewer"], "reviewer") + given = verdicts["steps"] + if not isinstance(given, dict): + raise ValueError("verdicts: steps must map step ids to verdicts") + if set(given) - set(shots): + raise ValueError("verdicts: unknown or skipped step ids") + reasons, results = [], {} + for identifier, step in shots.items(): + if digest(report_path.parent / step["screenshot"]) != step.get("screenshot_sha256"): + raise ValueError(f"screenshot changed since the run: {step['screenshot']}") + verdict = given.get(identifier) + if verdict is None: + reasons.append(f"missing_verdict:{identifier}") + continue + if ( + not isinstance(verdict, dict) + or set(verdict) != {"verdict", "note"} + or verdict["verdict"] not in VERDICTS + ): + raise ValueError(f"verdicts.{identifier}: verdict pass or fail and a note required") + _text(verdict["note"], f"verdicts.{identifier}.note") + results[identifier] = {**verdict, "screenshot_sha256": step["screenshot_sha256"]} + if verdict["verdict"] == "fail": + reasons.append(f"visual_fail:{identifier}") + if report["status"] != "passed": + reasons.append("automated_checks_failed") + if any(reason.startswith("missing_verdict") for reason in reasons): + status = "blocked" + else: + status = "failed" if reasons else "passed" + summary = { + "schema_version": 1, + "status": status, + "reasons": reasons, + "reviewer": verdicts["reviewer"], + "report_sha256": digest(report_path), + "steps": results, + "scope": "reviewer-declared visual verdicts; not authenticated by this tool", + } + (report_path.parent / "review.json").write_text(json.dumps(summary, indent=2) + "\n") + return summary + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + commands = parser.add_subparsers(dest="command", required=True) + runner = commands.add_parser("run", help="Run a UI script and save per-step evidence") + runner.add_argument("script") + runner.add_argument("--approved-origin", required=True) + runner.add_argument("--storage-state") + runner.add_argument("--out") + runner.add_argument("--headed", action="store_true") + runner.add_argument("--video", action="store_true") + runner.add_argument("--timeout-ms", type=int, default=15000) + reviewer = commands.add_parser("review", help="Record visual verdicts for a run") + reviewer.add_argument("report") + reviewer.add_argument("verdicts") + args = parser.parse_args(argv) + try: + if args.command == "run": + stamp = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%dT%H%M%SZ") + out = args.out or ROOT / "artifacts/sstack/ui" / f"{stamp}-{secrets.token_hex(4)}" + result = run( + json.loads(Path(args.script).read_text()), + approved_origin=args.approved_origin, + out_dir=out, + storage_state=args.storage_state, + headed=args.headed, + video=args.video, + timeout_ms=args.timeout_ms, + ) + result = {**result, "report": str(Path(out) / "report.json")} + else: + result = review(args.report, json.loads(Path(args.verdicts).read_text())) + except ImportError: + result = {"status": "blocked", "reasons": ["install playwright from requirements-dev"]} + except (OSError, ValueError) as error: + result = {"status": "blocked", "reasons": [str(error)]} + print(json.dumps(result, indent=2)) + return 0 if result.get("status") == "passed" else 1 + + +if __name__ == "__main__": + sys.exit(main())