From d9293bf36db01efb6398b2ed6cd6c802e83088e6 Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Mon, 19 Oct 2026 21:25:18 -0700 Subject: [PATCH 1/9] Keep parametrized test ids stable under pytest-xdist Two parametrized tests built their ids from uuid4(), so each xdist worker collected different ids and 'pytest -n 4' aborted at collection. Use fixed ids instead. --- tests/unit/operations/test_run_views.py | 5 +++-- tests/unit/test_ai_endpoints.py | 10 +++++++--- 2 files changed, 10 insertions(+), 5 deletions(-) diff --git a/tests/unit/operations/test_run_views.py b/tests/unit/operations/test_run_views.py index 9cd06724..52591329 100644 --- a/tests/unit/operations/test_run_views.py +++ b/tests/unit/operations/test_run_views.py @@ -19,7 +19,7 @@ import json from contextlib import asynccontextmanager from types import SimpleNamespace -from uuid import uuid4 +from uuid import UUID, uuid4 import pytest from starlette.requests import Request @@ -35,7 +35,8 @@ from firefly_weave.operations.run_views import RunViewService SCOPE = Scope(tenant_id=uuid4(), project_id=uuid4(), environment_id=uuid4()) -RUN = uuid4() +# A fixed id keeps parametrized test ids identical across pytest-xdist workers. +RUN = UUID("6f2c1a8e-3b4d-4e5f-9a7b-1c2d3e4f5a6b") CONTEXT = AuditContext() diff --git a/tests/unit/test_ai_endpoints.py b/tests/unit/test_ai_endpoints.py index 057fb281..8175e0ed 100644 --- a/tests/unit/test_ai_endpoints.py +++ b/tests/unit/test_ai_endpoints.py @@ -45,14 +45,18 @@ def test_model_query_is_required_and_defaults_remain_optional(): assert bindable["limit"].required is False and bindable["limit"].default == 50 +# A fixed id keeps parametrized test ids identical across pytest-xdist workers. +CONNECTION = "345e1767-4745-488b-b967-e3c4c07e62ac" + + @pytest.mark.parametrize( "query", [ "", "connection=wrong", - "connection=" + str(uuid4()) + "&refresh=wrong", - "connection=" + str(uuid4()) + "&refresh=true&refresh=false", - "connection=" + str(uuid4()) + "&unknown=1", + f"connection={CONNECTION}&refresh=wrong", + f"connection={CONNECTION}&refresh=true&refresh=false", + f"connection={CONNECTION}&unknown=1", ], ) async def test_model_queries_are_strict_before_service_access(query): From 369bf993979ba291ae0cdc9dcadb60311ea5c2a5 Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Mon, 19 Oct 2026 21:26:49 -0700 Subject: [PATCH 2/9] Draw decision table steps in workflow graph exports weave workflow graph raised KeyError: 'decisionTable' on any workflow with a decision table step because the graph labels had no entry for that node kind. Label it, and pin that every compiled node kind has a label. --- src/firefly_weave/sdk/visualization.py | 1 + tests/unit/cli/test_graph.py | 52 ++++++++++++++++++++++++++ 2 files changed, 53 insertions(+) diff --git a/src/firefly_weave/sdk/visualization.py b/src/firefly_weave/sdk/visualization.py index 593496c2..7d9a44f8 100644 --- a/src/firefly_weave/sdk/visualization.py +++ b/src/firefly_weave/sdk/visualization.py @@ -31,6 +31,7 @@ "end": "Return the result", "action": "Call an action", "transform": "Shape the data", + "decisionTable": "Apply a decision table", "wait": "Wait for a duration", "signal": "Wait for a signal", "humanTask": "Wait for a human decision", diff --git a/tests/unit/cli/test_graph.py b/tests/unit/cli/test_graph.py index b467bb67..24a0c71e 100644 --- a/tests/unit/cli/test_graph.py +++ b/tests/unit/cli/test_graph.py @@ -296,3 +296,55 @@ def test_graph_draws_a_compiled_call_without_running_it(tmp_path): assert "Call a workflow" in json.loads(drawn.output)["content"] text = CliRunner().invoke(cli, ["workflow", "graph", str(path)]).output assert "[notify] Call a workflow\n" in text and "[audit] Call a workflow\n" in text + + +@pytest.mark.parametrize("format", ["text", "mermaid", "svg"]) +def test_graph_draws_decision_table_steps(tmp_path, format): + from firefly_weave.compiler.catalog import CatalogSnapshot + from firefly_weave.contracts.definitions import load_definition + + schema = {"type": "object", "properties": {"amount": {"type": "number"}}} + table = { + "apiVersion": "weave/v1alpha1", + "kind": "DecisionTable", + "metadata": {"name": "payment-policy", "version": "1.0.0"}, + "spec": { + "inputSchema": schema, + "outputSchema": {"type": "string"}, + "hitPolicy": "first", + "rules": [{"id": "any", "when": {"literal": True}, "output": {"literal": "approve"}}], + }, + } + workflow = { + "apiVersion": "weave/v1alpha1", + "kind": "Workflow", + "metadata": {"name": "payment", "version": "1.0.0"}, + "spec": { + "inputSchema": schema, + "outputSchema": {"type": "string"}, + "steps": [ + {"id": "decide", "kind": "decisionTable", "uses": "payment-policy@1.0.0", "with": {"ref": "/input"}} + ], + "output": {"ref": "/steps/decide/output"}, + }, + } + catalog = CatalogSnapshot.from_definitions([load_definition(table)]) + flow = compile_source(json.dumps(workflow), format="json", catalog=catalog) + assert flow.artifact is not None, flow.to_bytes() + path = tmp_path / "compiled.json" + path.write_bytes(flow.artifact.to_bytes()) + result = CliRunner().invoke(cli, ["workflow", "graph", str(path), "--format", format, "--output", "json"]) + assert result.exit_code == 0, result.output + assert "Apply a decision table" in json.loads(result.output)["content"] + + +def test_graph_labels_every_compiled_node_kind(): + from typing import get_args + + from firefly_weave.compiler.ir import IRNode + from firefly_weave.sdk.visualization import LABELS + + union = get_args(IRNode.__value__)[0] + kinds = {kind for model in get_args(union) for kind in get_args(model.model_fields["kind"].annotation)} + assert "decisionTable" in kinds + assert kinds - set(LABELS) == set() From 464a9cf5e22d4128e7effcf4d1bd708916bcd25d Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Mon, 19 Oct 2026 21:27:54 -0700 Subject: [PATCH 3/9] Name the failing stage and tests when the gate fails scripts/check.py printed only 'Checks failed; inspect private evidence.', and CI keeps no evidence, so a failing unit test could not be identified. On failure it now prints the stage name and the FAILED/ERROR node ids from pytest's short summary. The ' - message' part, tracebacks and captured output stay in the private stage log. --- CONTRIBUTING.md | 6 ++- scripts/check.py | 28 +++++++++++ tests/unit/test_integration_fixture_runner.py | 47 +++++++++++++++++++ 3 files changed, 79 insertions(+), 2 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 1dde59c3..f128f80d 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -41,8 +41,10 @@ site build, Ruff lint and format checks, strict mypy, unit and contract tests, and release package preparation with installed-artifact verification. Expected: the last line is `All requested checks passed.` A failing stage stops -the run, which then prints `Checks failed; inspect private evidence.`; each -stage's log and a `checks.json` summary are in a new `build/checks-…` directory. +the run, which then prints `Stage failed.`, the node id of each failing +test from pytest's short summary (never its output), and +`Checks failed; inspect private evidence.`; each stage's log and a `checks.json` +summary are in a new `build/checks-…` directory. For a focused change, first run its relevant suite; the integrated delivery must still pass the full gate. Do not update lock or upstream framework provenance casually to resolve an unrelated diff --git a/scripts/check.py b/scripts/check.py index fe51cf86..4437f67c 100644 --- a/scripts/check.py +++ b/scripts/check.py @@ -21,6 +21,7 @@ import argparse import json import os +import re import runpy import signal import sys @@ -64,6 +65,30 @@ def worker_stages() -> list[tuple[str, list[str], int]]: return stages +# Only the outcome and node id of pytest's short summary lines; the " - message" part may carry output. +SUMMARY_LINE = re.compile(r"^(FAILED|ERROR) ([\x21-\x7e](?:[\x20-\x7e]*?[\x21-\x7e])?)(?: - |$)") +SUMMARY_TAIL_BYTES = 256 * 1024 +SUMMARY_MAX_LINES = 50 +SUMMARY_MAX_LENGTH = 300 + + +def failing_tests(log: Path) -> list[str]: + """Name the failing tests from the tail of a stage log, never their output.""" + try: + with log.open("rb") as stream: + stream.seek(max(0, stream.seek(0, os.SEEK_END) - SUMMARY_TAIL_BYTES)) + tail = stream.read().decode("utf-8", "replace") + except OSError: + return [] + _, header, summary = tail.rpartition("short test summary info") + names = [] + for line in summary.splitlines() if header else (): + match = SUMMARY_LINE.match(re.sub(r"\x1b\[[0-9;]*m", "", line)) + if match and len(match.group(2)) <= SUMMARY_MAX_LENGTH: + names.append(match.group(1) + " " + match.group(2)) + return list(dict.fromkeys(names))[:SUMMARY_MAX_LINES] + + def release_prerequisites(root: Path, context: str | None) -> None: if context != "colima-weave-tests" or os.environ.get("WEAVE_TEST_DOCKER_CONTEXT") != context: raise ValueError("An explicit owned release Docker context is required") @@ -215,6 +240,9 @@ def run(root: Path, evidence: Path, *, release: bool, context: str | None, integ except BaseException: results[name] = {"status": "failed", "seconds": round(time.monotonic() - started, 3)} failed = True + print(f"Stage {name} failed.", flush=True) + for line in failing_tests(evidence / (name + ".log")): + print(line, flush=True) break except BaseException: failed = True diff --git a/tests/unit/test_integration_fixture_runner.py b/tests/unit/test_integration_fixture_runner.py index 33bb91c9..e94e456f 100644 --- a/tests/unit/test_integration_fixture_runner.py +++ b/tests/unit/test_integration_fixture_runner.py @@ -229,3 +229,50 @@ def test_integration_has_bounded_full_suite_budget_and_actionable_failure_output assert "-v" in command and "--maxfail=1" in command assert "--tb=short" in command and "--show-capture=no" in command assert "--showlocals" not in command + + +def test_failed_pytest_stage_names_failing_tests_without_output(runner, monkeypatch, tmp_path, capsys): + evidence = tmp_path / "checks" + + def command(argv, **kwargs): + name = kwargs["log_path"].stem + kwargs["log_path"].write_bytes( + b"....F.E\n" + b"FAILED printed-by-a-test-before-the-summary\n" + b"E AssertionError: token=never-print-this-secret\n" + b"=========================== short test summary info ============================\n" + b"FAILED tests/unit/test_a.py::test_one[case-1] - AssertionError: token=never-print-this-secret\n" + b"ERROR tests/unit/test_b.py::test_two - RuntimeError: never-print-this-body\n" + b"FAILED tests/unit/test_c.py::test_three\x1b[31m\n" + b"1 failed, 1 error in 3.00s\n" + if name == "unit-contracts" + else b"" + ) + if name == "unit-contracts": + raise RuntimeError("controlled command failure") + + monkeypatch.setattr(runner, "run_command", command) + assert runner.run(ROOT, evidence, release=False, context=None) == 1 + printed = capsys.readouterr().out + assert "Stage unit-contracts failed" in printed + assert "FAILED tests/unit/test_a.py::test_one[case-1]\n" in printed + assert "ERROR tests/unit/test_b.py::test_two\n" in printed + assert "FAILED tests/unit/test_c.py::test_three\n" in printed + assert "printed-by-a-test" not in printed + assert "never-print" not in printed and "AssertionError" not in printed and "\x1b" not in printed + + +def test_failed_stage_without_pytest_summary_names_only_the_stage(runner, monkeypatch, tmp_path, capsys): + evidence = tmp_path / "checks" + + def command(argv, **kwargs): + name = kwargs["log_path"].stem + kwargs["log_path"].write_bytes(b"src/x.py:1:1: F401 secret-looking-output\n") + if name == "lint": + raise RuntimeError("controlled command failure") + + monkeypatch.setattr(runner, "run_command", command) + assert runner.run(ROOT, evidence, release=False, context=None) == 1 + printed = capsys.readouterr().out + assert "Stage lint failed" in printed + assert "secret-looking-output" not in printed and "F401" not in printed From 68a241dd9777fb595e4956d269deabff83d8c26d Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Mon, 19 Oct 2026 21:33:30 -0700 Subject: [PATCH 4/9] Send a capacity-refused change again only for the same person After a capacity refusal, Studio paused and sent the same keyed change again without rechecking who was signed in. If another person signed in during the pause, the host sent the replay as them. The replay now waits for any sign-in check still out, as the Workers and Runs bulk loops do, and is sent only when the person and workspace are unchanged; otherwise the refusal stands and nothing more is sent. --- studio/src/app/api.ts | 22 +++++++- studio/src/app/app.ts | 6 +++ studio/tests/api.test.ts | 51 +++++++++++++++++++ .../tests/browser/worker-operations.spec.ts | 48 +++++++++++++++++ 4 files changed, 126 insertions(+), 1 deletion(-) diff --git a/studio/src/app/api.ts b/studio/src/app/api.ts index c89dc1c2..44041744 100644 --- a/studio/src/app/api.ts +++ b/studio/src/app/api.ts @@ -127,6 +127,16 @@ export class StudioApi { }; /** Called when the local host no longer recognizes this window's pairing. */ onSessionEnded: (() => void) | null = null; + /** + * Who is signed in, and where, for a keyed change sent again after a + * capacity pause: the host sends it for whoever is signed in by then. + * `settled` waits for a sign-in check still out; `scope` names the person + * and workspace right then. The shell sets it. + */ + replayGuard: { + settled: () => Promise; + scope: () => string; + } | null = null; /** Pauses before replaying a request the host or platform turned away for capacity. */ wait = (milliseconds: number) => new Promise((resolve) => setTimeout(resolve, milliseconds)); @@ -167,7 +177,10 @@ export class StudioApi { h["X-Weave-CSRF"] = this.session.csrfToken; // A capacity or compatibility rejection means the host or platform did not admit the // request, so a read, or a change that carries an idempotency key, can be sent again. - const safe = method === "GET" || "Idempotency-Key" in h; + const keyed = method !== "GET" && "Idempotency-Key" in h; + const safe = method === "GET" || keyed; + // A change is sent again only for the person and workspace it was first sent for. + const scope = keyed ? this.replayGuard?.scope() : undefined; for (let attempt = 1; ; attempt++) { const scoped = signal ? cancellable(signal, timeout) : null; let response: Response; @@ -202,6 +215,13 @@ export class StudioApi { retryDelay(response.headers.get("Retry-After")), signal, ); + const guard = keyed ? this.replayGuard : null; + if (guard) { + await guard.settled(); + signal?.throwIfAborted(); + // Nothing ran: the refusal stands, rather than a send as someone else. + if (guard.scope() !== scope) throw error; + } continue; } if ( diff --git a/studio/src/app/app.ts b/studio/src/app/app.ts index b3f5616f..5cea63dc 100644 --- a/studio/src/app/app.ts +++ b/studio/src/app/app.ts @@ -1176,6 +1176,12 @@ export class App implements CanvasHost { ); }); this.api.onSessionEnded = () => this.sessionEnded(); + // A keyed change refused for capacity is sent again only once Studio knows + // who is signed in, and only for the person and workspace it was sent for. + this.api.replayGuard = { + settled: () => this.identitySettled(), + scope: () => scopeKeyOf(this), + }; this.loginWatcher.subscribe((event) => this.backgroundLogin(event)); afterEveryRender(() => { const title = this.pageTitle(); diff --git a/studio/tests/api.test.ts b/studio/tests/api.test.ts index 36c19b09..0c16b23b 100644 --- a/studio/tests/api.test.ts +++ b/studio/tests/api.test.ts @@ -252,6 +252,57 @@ describe("capacity rejections", () => { }); expect(other).toHaveBeenCalledTimes(1); }); + it("does not replay a keyed change once another person is signed in", async () => { + const fetch = vi.fn(async () => busy()); + vi.stubGlobal("fetch", fetch); + const client = api(); + let person = "jane"; + let settle: () => void = () => undefined; + const checked: string[] = []; + client.replayGuard = { + scope: () => person, + settled: () => + new Promise((resolve) => { + checked.push(person); + settle = resolve; + }), + }; + client.wait = vi.fn(async () => { + // The person changes while Studio pauses before the replay. + person = "john"; + }); + const pending = client.mutate( + "/studio/api/runs", + "POST", + {}, + undefined, + "k", + ); + await vi.waitFor(() => expect(checked).toEqual(["john"])); + // Nothing is sent while the sign-in check is still out. + expect(fetch).toHaveBeenCalledTimes(1); + settle(); + await expect(pending).rejects.toMatchObject({ + status: 429, + code: "WV-OPERATION-CAPACITY", + }); + expect(fetch).toHaveBeenCalledTimes(1); + }); + it("replays a keyed change after the sign-in check finds the same person", async () => { + const fetch = vi + .fn() + .mockResolvedValueOnce(busy()) + .mockResolvedValueOnce(ok()); + vi.stubGlobal("fetch", fetch); + const client = api(); + const settled = vi.fn(async () => undefined); + client.replayGuard = { scope: () => "jane", settled }; + await expect( + client.mutate("/studio/api/runs", "POST", {}, undefined, "k"), + ).resolves.toEqual({ items: [] }); + expect(settled).toHaveBeenCalledTimes(1); + expect(fetch).toHaveBeenCalledTimes(2); + }); it("does not replay other 429 answers", async () => { const fetch = vi.fn().mockResolvedValue(busy("WV-PAGE-LIMIT")); vi.stubGlobal("fetch", fetch); diff --git a/studio/tests/browser/worker-operations.spec.ts b/studio/tests/browser/worker-operations.spec.ts index 3e285726..434ccb4b 100644 --- a/studio/tests/browser/worker-operations.spec.ts +++ b/studio/tests/browser/worker-operations.spec.ts @@ -1521,6 +1521,54 @@ test.describe("another person on the same environment", () => { ).toHaveCount(0); }); + test("a change refused for capacity is not sent again once another person is signed in", async ({ + page, + }) => { + const people = await janeThenJohn(page); + const sent: { person: string; path: string }[] = []; + let reached: () => void = () => undefined; + const out = new Promise((resolve) => (reached = resolve)); + await page.route("**/workers/*/drain", async (r) => { + sent.push({ + person: people.person(), + path: new URL(r.request().url()).pathname + .split("/") + .slice(-2) + .join("/"), + }); + if (sent.length === 1) { + reached(); + // Studio pauses for Retry-After, then would send the same keyed change again. + return r.fulfill({ + status: 429, + headers: { "Retry-After": "2" }, + json: { code: "WV-OPERATION-CAPACITY", message: "Busy" }, + }); + } + return r.fulfill({ json: { revoked: false } }); + }); + await page.getByLabel("Select every worker on this list, up to 50").check(); + await page + .getByRole("region", { name: "Selected workers" }) + .getByRole("button", { name: "Drain" }) + .click(); + await page + .getByRole("dialog", { name: "Drain 3 workers?" }) + .getByRole("button", { name: "Drain" }) + .click(); + // The platform refuses the first drain; John signs in during the pause. + await out; + const refused = Date.now(); + await people.switch(); + expect(Date.now() - refused).toBeLessThan(2000); + await expect + .poll(() => people.lists.filter((item) => item.person === "john").length) + .toBeGreaterThan(0); + // Past the pause, nothing has gone out for John. + await page.waitForTimeout(Math.max(0, 3500 - (Date.now() - refused))); + expect(sent).toEqual([{ person: "jane", path: `${online}/drain` }]); + }); + test("a bulk action waits for the sign-in check before its next send", async ({ page, }) => { From c86fa1430e63f29973235083c53fc7dbbdde0fa9 Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Mon, 19 Oct 2026 21:43:40 -0700 Subject: [PATCH 5/9] Resend capacity-refused changes in the human-task race tests test_due_expiry_wins_concurrent_completion and test_completion_races_cancellation_under_run_lock failed with 429 WV-OPERATION-CAPACITY instead of 409 whenever one side held the project's operation lock longer than the other's 250 ms lock wait, which a cold or loaded database does. That depended on timing, not on which module ran first. A capacity refusal runs nothing, so the tests now send it again, as a client does, and assert the real outcome. A variant whose deadline tick holds the lock for 600 ms reproduces the 429 every time. --- tests/integration/test_human_tasks.py | 79 ++++++++++++++++++--------- 1 file changed, 54 insertions(+), 25 deletions(-) diff --git a/tests/integration/test_human_tasks.py b/tests/integration/test_human_tasks.py index 2acf0041..98245600 100644 --- a/tests/integration/test_human_tasks.py +++ b/tests/integration/test_human_tasks.py @@ -183,6 +183,20 @@ async def mutate(client, headers, env_url, task, command, payload, key): ) +async def admitted(send, attempts=20): + """Send again while the platform refuses for capacity, as a client does: such a refusal ran nothing. + + Concurrent changes in one project queue on its operation lock for at most 250 ms; a slower holder (a cold + or loaded database) makes the other side answer 429 WV-OPERATION-CAPACITY instead of its real outcome. + """ + for _ in range(attempts - 1): + response = await send() + if response.status_code != 429 or response.json().get("code") != "WV-OPERATION-CAPACITY": + return response + await asyncio.sleep(0.05) + return await send() + + @pytest.mark.parametrize("historical", [False, True]) async def test_claim_complete_retry_and_conflicting_payload( client, headers, env_url, human_run, access_db, author, historical @@ -374,16 +388,18 @@ async def test_completion_races_cancellation_under_run_lock(client, headers, env run, task = human_run assert (await mutate(client, headers, env_url, task, "claim", {"expected_revision": 1}, "claim")).status_code == 200 complete, cancel = await asyncio.gather( - mutate( - client, - headers, - env_url, - task, - "complete", - {"expected_revision": 2, "decision": "approve", "data": {"note": "ok"}}, - "complete", + admitted( + lambda: mutate( + client, + headers, + env_url, + task, + "complete", + {"expected_revision": 2, "decision": "approve", "data": {"note": "ok"}}, + "complete", + ) ), - client.post(f"{env_url}/runs/{run}/cancel", headers=headers, json={"reason": "Cancel"}), + admitted(lambda: client.post(f"{env_url}/runs/{run}/cancel", headers=headers, json={"reason": "Cancel"})), ) assert sorted((complete.status_code, cancel.status_code)) == [200, 409], [complete.text, cancel.text] state = (await client.get(f"{env_url}/runs/{run}", headers=headers)).json()["state"] @@ -393,7 +409,10 @@ async def test_completion_races_cancellation_under_run_lock(client, headers, env assert count == (1 if state["status"] == "succeeded" else 0) -async def test_due_expiry_wins_concurrent_completion(client, headers, env_url, human_run, access_db, services, author): +@pytest.mark.parametrize("held", [False, True], ids=["tick", "tick-holding-the-project-lock"]) +async def test_due_expiry_wins_concurrent_completion( + client, headers, env_url, human_run, access_db, services, author, held +): from firefly_weave.persistence.uow import UnitOfWork from firefly_weave.runtime.deadlines import DeadlineService @@ -401,24 +420,34 @@ async def test_due_expiry_wins_concurrent_completion(client, headers, env_url, h assert (await mutate(client, headers, env_url, task, "claim", {"expected_revision": 1}, "claim")).status_code == 200 async with access_db[1].begin() as session: await session.execute(text("UPDATE run_deadlines SET deadline=clock_timestamp() WHERE node_id='review'")) + locked = asyncio.Event() async def expire(): async with UnitOfWork(access_db[0]).open(author[2]) as tx: - return await services(access_db[0]).resolve(DeadlineService).tick(tx, 100) - - complete, _ = await asyncio.gather( - mutate( - client, - headers, - env_url, - task, - "complete", - {"expected_revision": 2, "decision": "approve", "data": {"note": "ok"}}, - "complete", - ), - expire(), - ) - assert complete.status_code == 409, complete.text + locked.set() + result = await services(access_db[0]).resolve(DeadlineService).tick(tx, 100) + if held: + # A slow tick keeps the project's operation lock past a request's 250 ms lock wait. + await tx.session.execute(text("SELECT pg_sleep(0.6)")) + return result + + async def complete(): + if held: + await locked.wait() + return await admitted( + lambda: mutate( + client, + headers, + env_url, + task, + "complete", + {"expected_revision": 2, "decision": "approve", "data": {"note": "ok"}}, + "complete", + ) + ) + + completed, _ = await asyncio.gather(complete(), expire()) + assert completed.status_code == 409, completed.text assert (await client.get(f"{env_url}/runs/{run}", headers=headers)).json()["state"]["status"] == "timed_out" async with access_db[1]() as session: assert await session.scalar(text("SELECT count(*) FROM human_task_decisions")) == 0 From 04e54740697add51922d7fc589d8ea17a9cb23dd Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Mon, 19 Oct 2026 21:47:16 -0700 Subject: [PATCH 6/9] Grant viewer on the demo project in weave platform user weave platform user granted viewer on the demo environment, but the catalog is a project resource and an environment grant never contains a project-level read (tenant > project > environment). A person created with operator or deployment_approver plus viewer could not see published workflows in Studio, although the guide says viewer reads the catalog. viewer is now granted on the demo project, as developer is. --- docs/guides/local-platform.md | 2 +- docs/guides/people-and-access.md | 2 +- docs/guides/roles-and-lifecycle.md | 4 +-- docs/reference/cli.md | 10 ++++--- src/firefly_weave/cli/platform.py | 6 ++-- src/firefly_weave/sdk/platform.py | 6 ++-- tests/unit/cli/test_platform_sign_in.py | 37 +++++++++++++++++++++++-- 7 files changed, 52 insertions(+), 15 deletions(-) diff --git a/docs/guides/local-platform.md b/docs/guides/local-platform.md index 48b76fbe..6846e216 100644 --- a/docs/guides/local-platform.md +++ b/docs/guides/local-platform.md @@ -234,7 +234,7 @@ read runs, and work on human tasks: | `developer` | Demo project | Read the catalog, compile, simulate, write, and publish definitions | | `deployer` | Demo environment | Activate and retire versions, bind connections, manage triggers and subscriptions | | `operator` | Demo environment | Start, pause, resume, retry, cancel, and signal runs; inspect and resolve incidents | -| `viewer` | Demo environment | Read runs, deliveries, status, and the catalog | +| `viewer` | Demo project | Read the catalog, and runs, deliveries, and status in the project's environments | | `task_participant` | Demo environment | Read, claim, complete, and release assigned human tasks | **`user` never changes an existing username**, and `--role` replaces the diff --git a/docs/guides/people-and-access.md b/docs/guides/people-and-access.md index 7916f445..83e23e6e 100644 --- a/docs/guides/people-and-access.md +++ b/docs/guides/people-and-access.md @@ -400,7 +400,7 @@ weave platform user --username alice Expected: `Username: alice`, the password, the granted roles, and the next command, `weave auth setup http://127.0.0.1:API_PORT`. The default roles are -`developer` on the demo project, and `deployer`, `operator`, `viewer`, and +`developer` and `viewer` on the demo project, and `deployer`, `operator`, and `task_participant` on the demo environment. Repeat `--role` to grant a different set instead. An existing username is never reset. Never use this command with a shared or production identity provider. diff --git a/docs/guides/roles-and-lifecycle.md b/docs/guides/roles-and-lifecycle.md index 61f265f2..9fbfc618 100644 --- a/docs/guides/roles-and-lifecycle.md +++ b/docs/guides/roles-and-lifecycle.md @@ -87,8 +87,8 @@ an author cannot start runs unless you grant that too. | Operator | Inspect, pause, resume, retry, or cancel runs; resolve incidents; maintain the platform | `operator` and `viewer`; `execution_manager` to permanently delete run content | [Execution management](execution-management.md) and [incident operations](../reference/incident-operations.md) | On a local platform, `weave platform user` creates a development person with a -practical starting set: `developer` on the demo project, and `deployer`, -`operator`, `viewer`, and `task_participant` on the demo environment. That +practical starting set: `developer` and `viewer` on the demo project, and +`deployer`, `operator`, and `task_participant` on the demo environment. That person can author, publish, activate, run, inspect runs, and complete tasks, but cannot create integration connections without `tenant_admin`. See [Create a person who can sign in](local-platform.md#5-create-a-person-who-can-sign-in) diff --git a/docs/reference/cli.md b/docs/reference/cli.md index 2d93be7f..20f7659b 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -398,12 +398,14 @@ people connect with `_`, or `-`. - The generated password is printed once and never saved, and an existing username is never reset. -- Without `--role`, the person gets `developer` on the demo project and - `deployer`, `operator`, `viewer`, and `task_participant` on the demo +- Without `--role`, the person gets `developer` and `viewer` on the demo + project and `deployer`, `operator`, and `task_participant` on the demo environment. - Each `--role` replaces those defaults: `tenant_admin` is granted on the - tenant, `developer` on the project, and every other role on the environment. - Creating connections needs `tenant_admin`. + tenant, `developer` and `viewer` on the project, and every other role on the + environment. Published workflows are project resources, so a person without + `developer` needs `viewer` to see them. Creating connections needs + `tenant_admin`. **`integrations` and `secret` are for development only.** diff --git a/src/firefly_weave/cli/platform.py b/src/firefly_weave/cli/platform.py index 493c1c08..6f76b8a9 100644 --- a/src/firefly_weave/cli/platform.py +++ b/src/firefly_weave/cli/platform.py @@ -269,8 +269,10 @@ def user(context: click.Context, username: str | None, roles: tuple[str, ...], o never saved; an existing username is never reset. Default roles are the smallest set to author, publish, activate, run, - inspect runs, and work on human tasks: developer on the demo project, and - deployer, operator, viewer, and task_participant on the demo environment. + inspect runs, and work on human tasks: developer and viewer on the demo + project, and deployer, operator, and task_participant on the demo + environment. Add viewer to any other set so the person can read the + project's published workflows. Use `user grant` to add roles to an account later. """ if context.invoked_subcommand is not None: diff --git a/src/firefly_weave/sdk/platform.py b/src/firefly_weave/sdk/platform.py index 5344aa7b..30b60a10 100644 --- a/src/firefly_weave/sdk/platform.py +++ b/src/firefly_weave/sdk/platform.py @@ -1548,8 +1548,10 @@ def _person_roles(roles: Sequence[str], default: Sequence[str] = DEFAULT_PERSON_ def _person_grants(roles: tuple[str, ...], scope: dict[str, str]) -> list[dict[str, Any]]: - levels = {"tenant_admin": ("tenant_id",), "developer": ("tenant_id", "project_id")} - # Definitions are project resources; activation, runs, and tasks belong to the demo environment. + project = ("tenant_id", "project_id") + levels = {"tenant_admin": ("tenant_id",), "developer": project, "viewer": project} + # Definitions are project resources, and an environment grant never contains a project-level read: viewer + # reads the catalog only on the project. Activation, runs, and tasks belong to the demo environment. return [ { "role": role, diff --git a/tests/unit/cli/test_platform_sign_in.py b/tests/unit/cli/test_platform_sign_in.py index 2274833f..ab5e599d 100644 --- a/tests/unit/cli/test_platform_sign_in.py +++ b/tests/unit/cli/test_platform_sign_in.py @@ -438,7 +438,7 @@ def test_user_creates_account_principal_link_and_default_grants(demo_workspace, "principal_id": PRINCIPAL, "role": role, "project_id": PROJECT, - "environment_id": ENVIRONMENT, + "environment_id": None if role == "viewer" else ENVIRONMENT, "resources": [], } for role in ("deployer", "operator", "viewer", "task_participant") @@ -459,7 +459,7 @@ def test_user_creates_account_principal_link_and_default_grants(demo_workspace, @pytest.mark.parametrize( ("roles", "expected"), [ - (("viewer",), [("viewer", PROJECT, ENVIRONMENT)]), + (("viewer",), [("viewer", PROJECT, None)]), (("developer",), [("developer", PROJECT, None)]), (("tenant_admin", "task_manager"), [("tenant_admin", None, None), ("task_manager", PROJECT, ENVIRONMENT)]), ], @@ -710,7 +710,7 @@ def test_cli_text_shows_password_once_with_warning_and_next_commands(demo_worksp assert "Username: grace" in lines assert any("Development only" in line and "not saved" in line for line in lines) assert " weave auth setup " + API in lines and " weave studio" in lines - assert "Roles: viewer (demo environment)" in lines + assert "Roles: viewer (demo project)" in lines def test_cli_reports_existing_username_as_platform_error(demo_workspace, monkeypatch): @@ -958,3 +958,34 @@ def test_cli_text_says_when_the_initial_account_receipt_was_not_completed(capsys output = capsys.readouterr().out assert output.count("shown-once") == 1 assert platform._RECEIPT_NOT_SAVED in output + + +@pytest.mark.parametrize( + "roles", + [ + ("operator", "viewer"), + ("deployment_approver", "viewer"), + ("deployer", "operator", "viewer", "task_participant"), + platform.DEFAULT_PERSON_ROLES, + ], +) +def test_people_with_viewer_read_the_project_catalog_and_their_environment(roles): + # The catalog is a project resource; an environment grant never contains a project-level read. + from firefly_weave.access.authorization import AuthorizationService + from firefly_weave.access.models import Grant, Principal + from firefly_weave.contracts.access import Scope + + workspace = {"tenant_id": TENANT, "project_id": PROJECT, "environment_id": ENVIRONMENT} + grants = [ + Grant(role=item["role"], scope=Scope(**item["scope"])) + for item in platform._person_grants(tuple(roles), workspace) + ] + person = Principal(id=uuid4(), kind="human", active=True, grants=grants) + project = Scope(tenant_id=UUID(TENANT), project_id=UUID(PROJECT)) + environment = Scope(tenant_id=UUID(TENANT), project_id=UUID(PROJECT), environment_id=UUID(ENVIRONMENT)) + access = AuthorizationService() + assert access.allows(person, project, "catalog.read") + assert access.allows(person, environment, "catalog.read") + assert access.allows(person, environment, "run.read") + other_project = Scope(tenant_id=UUID(TENANT), project_id=uuid4()) + assert not access.allows(person, other_project, "catalog.read") From 8009e6dd709545b0a70fd226cf6e0ec2d40eb7b2 Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Mon, 19 Oct 2026 21:47:24 -0700 Subject: [PATCH 7/9] Note the graph, Studio replay, platform user and check fixes in the changelog --- CHANGELOG.md | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index eb79d718..fdd26b49 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -462,6 +462,20 @@ SPDX-License-Identifier: Apache-2.0 - While the platform is not ready, the outbox waits for readiness instead of starting a delivery turn every second that is refused and logged as `Outbox turn failed` at ERROR. +- `weave workflow graph` draws decision table steps as **Apply a decision + table**. Before, it stopped with `KeyError: 'decisionTable'` on any workflow + with a decision table. +- Studio sends a change that the platform refused for capacity again only when + the same person is still signed in to the same workspace. It first waits for + a sign-in check still out. Before, a change refused just as another person + signed in could be sent again as that person. +- `weave platform user` grants `viewer` on the demo project instead of the demo + environment, so a person created with roles such as `operator` or + `deployment_approver` plus `viewer` can read the project's published + workflows. +- When `make check` fails, it prints the failing stage and the node ids of the + failing tests from pytest's short summary. Test output stays in the private + stage logs. ## 0.1.0a15 From 90a15586fd08511ccbb2a6fdec81519fc49536b3 Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Mon, 19 Oct 2026 21:57:01 -0700 Subject: [PATCH 8/9] Resend capacity-refused changes in the other human-task race tests The concurrent claim, concurrent completion and keyed run start tests race two changes in one project the same way, and the completion race failed with 429 WV-OPERATION-CAPACITY in a tasks-before-files run. --- tests/integration/test_human_tasks.py | 32 ++++++++++++++++++--------- 1 file changed, 21 insertions(+), 11 deletions(-) diff --git a/tests/integration/test_human_tasks.py b/tests/integration/test_human_tasks.py index 98245600..957529bd 100644 --- a/tests/integration/test_human_tasks.py +++ b/tests/integration/test_human_tasks.py @@ -275,7 +275,12 @@ async def test_pause_allows_fact_then_resume_continues(client, headers, env_url, async def test_concurrent_claim_accepts_once(client, headers, env_url, human_run): _, task = human_run responses = await asyncio.gather( - *(mutate(client, headers, env_url, task, "claim", {"expected_revision": 1}, f"claim-{i}") for i in range(2)) + *( + admitted( + lambda i=i: mutate(client, headers, env_url, task, "claim", {"expected_revision": 1}, f"claim-{i}") + ) + for i in range(2) + ) ) assert sorted(r.status_code for r in responses) == [200, 409], [r.text for r in responses] @@ -377,7 +382,10 @@ async def test_concurrent_complete_records_one_immutable_decision(client, header assert (await mutate(client, headers, env_url, task, "claim", {"expected_revision": 1}, "claim")).status_code == 200 payload = {"expected_revision": 2, "decision": "approve", "data": {"note": "ok"}} responses = await asyncio.gather( - *(mutate(client, headers, env_url, task, "complete", payload, f"complete-{i}") for i in range(2)) + *( + admitted(lambda i=i: mutate(client, headers, env_url, task, "complete", payload, f"complete-{i}")) + for i in range(2) + ) ) assert sorted(r.status_code for r in responses) == [200, 409], [r.text for r in responses] async with access_db[1]() as session: @@ -460,15 +468,17 @@ async def test_run_keys_filter_distinct_executions_and_bind_cursor(human_run, au activation = existing.json()["activation"]["id"] responses = await asyncio.gather( *[ - client.post( - env_url + "/runs", - headers={**headers, "Idempotency-Key": "keyed-" + str(i)}, - json={ - "activation_id": activation, - "input": {}, - "business_key": "expense-42", - "correlation_key": "thread-7", - }, + admitted( + lambda i=i: client.post( + env_url + "/runs", + headers={**headers, "Idempotency-Key": "keyed-" + str(i)}, + json={ + "activation_id": activation, + "input": {}, + "business_key": "expense-42", + "correlation_key": "thread-7", + }, + ) ) for i in range(2) ] From d98f1920ac9f14c6e9485f0e3599207bcf575317 Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Mon, 19 Oct 2026 22:42:46 -0700 Subject: [PATCH 9/9] Let a cancel end the sign-in wait and read only pytest's summary From the independent review: a canceled request no longer waits for a slow sign-in check before it stops, the gate reads node ids only up to pytest's closing summary line, and the browser test of the capacity replay leaves a slow runner more time inside the pause. --- scripts/check.py | 5 +++- studio/src/app/api.ts | 13 +++++---- studio/tests/api.test.ts | 27 +++++++++++++++++++ .../tests/browser/worker-operations.spec.ts | 7 ++--- tests/unit/test_integration_fixture_runner.py | 5 ++-- 5 files changed, 46 insertions(+), 11 deletions(-) diff --git a/scripts/check.py b/scripts/check.py index 4437f67c..8475bce1 100644 --- a/scripts/check.py +++ b/scripts/check.py @@ -82,7 +82,10 @@ def failing_tests(log: Path) -> list[str]: return [] _, header, summary = tail.rpartition("short test summary info") names = [] - for line in summary.splitlines() if header else (): + for line in summary.splitlines()[1:] if header else (): + # pytest closes its summary with a "==== 1 failed in 3.00s ====" line; later output is not its summary. + if line.startswith("="): + break match = SUMMARY_LINE.match(re.sub(r"\x1b\[[0-9;]*m", "", line)) if match and len(match.group(2)) <= SUMMARY_MAX_LENGTH: names.append(match.group(1) + " " + match.group(2)) diff --git a/studio/src/app/api.ts b/studio/src/app/api.ts index 44041744..e4af82de 100644 --- a/studio/src/app/api.ts +++ b/studio/src/app/api.ts @@ -141,8 +141,12 @@ export class StudioApi { wait = (milliseconds: number) => new Promise((resolve) => setTimeout(resolve, milliseconds)); /** A pause that a cancel ends at once, so a canceled request is never sent again. */ - private async pause(milliseconds: number, signal?: AbortSignal) { - if (!signal) return this.wait(milliseconds); + private pause(milliseconds: number, signal?: AbortSignal) { + return this.untilCanceled(() => this.wait(milliseconds), signal); + } + /** Waits for `work`, or rejects at once when `signal` cancels first. */ + private async untilCanceled(work: () => Promise, signal?: AbortSignal) { + if (!signal) return work(); signal.throwIfAborted(); let stop = () => {}; const canceled = new Promise((_resolve, reject) => { @@ -150,7 +154,7 @@ export class StudioApi { signal.addEventListener("abort", stop, { once: true }); }); try { - await Promise.race([this.wait(milliseconds), canceled]); + await Promise.race([work(), canceled]); } finally { signal.removeEventListener("abort", stop); } @@ -217,8 +221,7 @@ export class StudioApi { ); const guard = keyed ? this.replayGuard : null; if (guard) { - await guard.settled(); - signal?.throwIfAborted(); + await this.untilCanceled(() => guard.settled(), signal); // Nothing ran: the refusal stands, rather than a send as someone else. if (guard.scope() !== scope) throw error; } diff --git a/studio/tests/api.test.ts b/studio/tests/api.test.ts index 0c16b23b..4086a94a 100644 --- a/studio/tests/api.test.ts +++ b/studio/tests/api.test.ts @@ -288,6 +288,33 @@ describe("capacity rejections", () => { }); expect(fetch).toHaveBeenCalledTimes(1); }); + it("stops at once when canceled while the sign-in check is still out", async () => { + const fetch = vi.fn(async () => busy()); + vi.stubGlobal("fetch", fetch); + const client = api(); + let waiting: () => void = () => undefined; + const reached = new Promise((resolve) => (waiting = resolve)); + client.replayGuard = { + scope: () => "jane", + settled: () => { + waiting(); + return new Promise(() => undefined); + }, + }; + const controller = new AbortController(); + const pending = client.request( + "/studio/api/runs", + "POST", + {}, + { "Idempotency-Key": "k" }, + 15000, + controller.signal, + ); + await reached; + controller.abort(new DOMException("Canceled.", "AbortError")); + await expect(pending).rejects.toMatchObject({ name: "AbortError" }); + expect(fetch).toHaveBeenCalledTimes(1); + }); it("replays a keyed change after the sign-in check finds the same person", async () => { const fetch = vi .fn() diff --git a/studio/tests/browser/worker-operations.spec.ts b/studio/tests/browser/worker-operations.spec.ts index 434ccb4b..fd460d81 100644 --- a/studio/tests/browser/worker-operations.spec.ts +++ b/studio/tests/browser/worker-operations.spec.ts @@ -1541,7 +1541,7 @@ test.describe("another person on the same environment", () => { // Studio pauses for Retry-After, then would send the same keyed change again. return r.fulfill({ status: 429, - headers: { "Retry-After": "2" }, + headers: { "Retry-After": "5" }, json: { code: "WV-OPERATION-CAPACITY", message: "Busy" }, }); } @@ -1560,12 +1560,13 @@ test.describe("another person on the same environment", () => { await out; const refused = Date.now(); await people.switch(); - expect(Date.now() - refused).toBeLessThan(2000); + // The switch lands inside the five-second pause, even on a slow runner. + expect(Date.now() - refused).toBeLessThan(4500); await expect .poll(() => people.lists.filter((item) => item.person === "john").length) .toBeGreaterThan(0); // Past the pause, nothing has gone out for John. - await page.waitForTimeout(Math.max(0, 3500 - (Date.now() - refused))); + await page.waitForTimeout(Math.max(0, 6500 - (Date.now() - refused))); expect(sent).toEqual([{ person: "jane", path: `${online}/drain` }]); }); diff --git a/tests/unit/test_integration_fixture_runner.py b/tests/unit/test_integration_fixture_runner.py index e94e456f..41278add 100644 --- a/tests/unit/test_integration_fixture_runner.py +++ b/tests/unit/test_integration_fixture_runner.py @@ -244,7 +244,8 @@ def command(argv, **kwargs): b"FAILED tests/unit/test_a.py::test_one[case-1] - AssertionError: token=never-print-this-secret\n" b"ERROR tests/unit/test_b.py::test_two - RuntimeError: never-print-this-body\n" b"FAILED tests/unit/test_c.py::test_three\x1b[31m\n" - b"1 failed, 1 error in 3.00s\n" + b"===================== 2 failed, 1 error in 3.00s ======================\n" + b"FAILED printed-after-the-summary\n" if name == "unit-contracts" else b"" ) @@ -258,7 +259,7 @@ def command(argv, **kwargs): assert "FAILED tests/unit/test_a.py::test_one[case-1]\n" in printed assert "ERROR tests/unit/test_b.py::test_two\n" in printed assert "FAILED tests/unit/test_c.py::test_three\n" in printed - assert "printed-by-a-test" not in printed + assert "printed-by-a-test" not in printed and "printed-after-the-summary" not in printed assert "never-print" not in printed and "AssertionError" not in printed and "\x1b" not in printed