From 27cb73659d8af7acfdc03b2b770dcee85f2d60fb Mon Sep 17 00:00:00 2001 From: Otavio Rodrigues Santana Date: Wed, 2 Sep 2026 21:17:26 -0300 Subject: [PATCH] fix(auth): correct OAuth conflict error message + validate multi-worker Postgres claim with real concurrency benchmark (#5026) * fix(auth): correct OAuth uniqueness error and index parity on Postgres create_user() caught any IntegrityError on commit and always reported it as a duplicate email. The email pre-check already rules out a real email collision in the common case, so any IntegrityError reaching that handler is actually idx_users_oauth_identity firing instead -- confirmed against both backends: SQLite reports "UNIQUE constraint failed: users.oauth_provider, users.oauth_id", Postgres reports a UniqueViolationError naming the same index. The caller saw "Email already registered" for an OAuth account conflict, which is wrong and would send API consumers debugging the wrong field. Distinguish the two cases via a substring check on the driver error text (both backends name the oauth columns) and raise an accurate message for each. Also add postgresql_where to the same index, alongside the existing sqlite_where. This is not a correctness fix -- verified empirically that Postgres already enforces the same practical uniqueness without it (NULL is never equal to NULL in either backends unique index, so real duplicate (provider, id) pairs are already rejected and NULL/NULL rows are already unconstrained). postgresql_where makes the index genuinely partial on Postgres too, matching the stated intent in the surrounding comment and keeping the index smaller as the common case (plain-password accounts, both columns NULL) accumulates. * test(bench): add multi-process SQLite vs Postgres concurrency benchmark CONFIGURATION.md documents that multi-worker deployments must use Postgres because "SQLite silently ignores row-level locks", but nothing in the repo exercised that claim against real separate worker processes -- the existing checkpoint benchmarks (scripts/benchmark/checkpoint/) measure single-process read/write latency, and the existing Postgres tests (test_pg_schema_integration.py, test_multi_worker_postgres_gate.py) cover schema placement and config validation, not throughput or lock behavior under concurrent load. run_concurrency_bench.py spawns N real OS processes (subprocess.Popen, not asyncio tasks or threads within one process) against the shared users table, mixing reads (get_user_by_email) and writes (create_user) at a configurable ratio, and reports throughput, error counts by exception type, and p50/p95/p99/max latency per run. Measured locally (2/4/8/16 workers, 100 ops/worker, 70/30 read/write): SQLite completed all operations with zero errors at every worker count (busy_timeout absorbs contention rather than raising), but total throughput stayed flat around 28-34 ops/s regardless of worker count, and p99 latency grew from ~400ms at 2 workers to ~5.9s at 16, with a 22s max. Postgres throughput scaled with worker count (41 to 66 ops/s) and p99 stayed under 500ms at every worker count tested. Raw JSON output from both runs is available on request; exact numbers will vary by machine and are not asserted in the test suite. test_bench_concurrency.py unit-tests the pure aggregation logic (percentile math, error grouping, crashed-worker handling) the same way test_bench_checkpoint_channels.py does for the existing benchmarks -- fast, no DB required, not the full multi-process sweep in CI. * fix(auth): inspect the driver exception for OAuth conflict detection str(exc) embeds the full failed INSERT statement, whose column list always names oauth_provider/oauth_id, so a substring check on it misclassified every commit-time IntegrityError on the users table as an OAuth conflict (reproduced on SQLite: a duplicate primary key with a different email raised "OAuth account already linked: None/None"). _is_oauth_identity_violation now inspects exc.orig instead: constraint_name on Postgres, both violated column names present (not a bare "oauth" substring) on SQLite. Also ships the alembic revision idx_users_oauth_identity's postgresql_where predicate needed: 0001_baseline created it as a full index on Postgres, and ORM metadata changes only affect fresh create_all databases, never an already-versioned deployment. Addresses review feedback from willem-bd. * fix(bench): run the concurrency benchmark in an isolated schema and derive paths from the checkout --pg-url accepted an arbitrary database URL while the code pinned postgres_schema="public" and unconditionally ran DELETE FROM users -- against any non-disposable database that permanently destroyed every auth account. Each run now generates a unique throwaway schema (bench_), points both the seeder and every worker subprocess at it via postgres_schema, and drops only that schema (DROP SCHEMA ... CASCADE) once the full worker-count sweep finishes. Also stopped hard-coding /opt/deer-flow/backend as the checkout path and .venv/bin/python3 as the interpreter: BACKEND_DIR is now derived from Path(__file__), and workers are spawned with sys.executable (the orchestrator's own interpreter) instead, so the documented uv run python scripts/benchmark/concurrency/run_concurrency_bench.py command works from any checkout. Addresses review feedback from willem-bd. * fix: shorten oauth-index revision id, repin migration-head assertions, fix bench read/write mix - 0017_users_oauth_identity_partial_pg (36 chars) exceeded alembic_version.version_num's VARCHAR(32) limit, which would fail stamping/upgrading on both fresh and existing Postgres deployments. Renamed to 0017_oauth_identity_pg_partial (30 chars). - Repinned every test asserting 0016_subagent_batches as the migration head (test_persistence_bootstrap[.py|_concurrency.py|_regression.py], test_migration_0004/0007/0015) to the new 0017 revision id. - worker.py's `(i % 100) < int(read_ratio * 100)` assumed n_ops >= 100; at the documented default (50 ops/worker, 0.7 read ratio) it produced either all-reads or all-writes, never the claimed mixed workload. Replaced with read_count()/is_read_op(), which distribute an exact round(n_ops * read_ratio) reads evenly across the sequence via modular spacing, and added test_bench_worker.py covering the default values plus small op counts. * fix(bench): establish a real physical connection before timing ops async with sf(): pass entered an empty AsyncSession without checking out a physical connection -- SQLAlchemy stays lazy until the first statement executes. That pushed connection-establishment cost onto each worker's first timed operation instead of conn_time_s, and at 16 workers those 16 cold first-ops (1% of a 1600-op sample) could skew the reported p99. Execute a real `SELECT 1` before starting the timer instead. Verified with a real end-to-end run (uv sync + sqlite backend, 2 workers/10 ops, 0 errors) plus the full auth/bench/migration-bootstrap suites (135 tests) and ruff check/format, all clean. * fix(bench): synchronize workers before timing, fix percentile off-by-one Two remaining measurement issues from review: - run_workers() started the wall clock before spawning any worker, so throughput/wall_time absorbed N processes' staggered Python-startup and connection-establishment cost, and early workers could run ahead of ones still starting. Workers now print READY right before their timed loop and block on stdin for a GO signal; the orchestrator waits for every READY, then starts the timer and releases all workers together. - summarize()'s pct() used int(len(latencies) * p) directly as a zero-based index -- a one-based-rank-as-index bug that put p95 and p99 at the same slot (the max) for any 20-or-fewer-sample run, and for the documented 100-sample default. Now delegates to checkpoint_bench_common.percentile(), the already-correct nearest-rank implementation used elsewhere in the same benchmark family, instead of a second, broken one. Verified: 14/14 unit tests pass (2 new pinned-value regression tests for the percentile bug, using the reviewer's own 20-sample repro), ruff clean, and a real 2/4-worker SQLite multi-process smoke run completes with distinct p95/p99/max latencies and no hang. * fix(bench): absolute SQLite bench path, surface crash diagnostics, exit nonzero on failure; share OAuth index constant + cover Postgres branch Three more findings from review at 5fd25a7: - seed_baseline() cleaned an absolute .deer-flow/bench_data path but handed DatabaseConfig a relative one, which resolves against the CALLER's CWD -- not BACKEND_DIR. Invoking the documented command from anywhere other than backend/ silently pointed the seeder and the (cwd=BACKEND_DIR) workers at two different directories: workers crashed with 'unable to open database file' while the run still printed a well-formed summary and exited 0. Both seed_baseline() and worker.py's make_session_factory() now use the same absolute path. - Crashed workers' stderr was captured then discarded, and main() always exited 0 -- an all-crashed sweep was indistinguishable from a real (uneventful) measurement to anything checking the exit code or --out. run_workers() now prints each crash immediately and tags it with the real worker_id (previously always None); summarize() exposes crashed_worker_errors alongside the existing crashed_workers count; main() exits 1 via the new summary_indicates_failure() whenever any sweep crashed or fell short of expected_total_ops. - idx_users_oauth_identity was hardcoded separately in the ORM Index and in _is_oauth_identity_violation's Postgres branch, with no test to catch drift, and that branch had zero non-skipped coverage (its only guard needs a live Postgres CI never configures). Exported OAUTH_IDENTITY_INDEX_NAME from user/model.py as the shared source of truth (migrations intentionally keep their own frozen literal, matching every other revision in that package) and added stub-exception unit tests pinning both the asyncpg constraint_name path and the sqlite message-substring path, positive and negative. Verified: 107 passed locally (auth + bench-unit suites), ruff clean, and two real reproductions -- invoking run_concurrency_bench.py from a scratch directory outside backend/ (the reviewer's exact repro) now completes 8/8 ops with crashed_workers: 0 instead of crashing, and the new crashed_worker_errors/exit-code logic is exercised directly by the new unit tests against the real summarize()/summary_indicates_failure(). * fix(auth): attribute create_user IntegrityErrors to the right constraint Two coupled review findings on the classification helpers: P3 (fall-through) -- after ruling out the OAuth-identity index, create_user raised "Email already registered: {email}" for every remaining IntegrityError, including the duplicate-primary-key case the new regression test exercises, whose address is not registered. Added _is_email_violation() so the email message is used only for an actual users.email collision that raced past the pre-check; anything else (in practice a duplicate id) now raises a neutral "User already exists (constraint: )". P2 (unreachable asyncpg branch) -- exc.orig is not the asyncpg error. SQLAlchemy's asyncpg dialect re-raises its own DBAPI IntegrityError (pgcode/sqlstate only) 'from' the real asyncpg error, so constraint_name lives on exc.orig.__cause__. getattr(exc.orig, "constraint_name", None) was always None on Postgres; the helpers only worked there by accident, matching asyncpg's DETAIL line in the message fallback. Added _driver_constraint_name() which walks orig then orig.__cause__, and the stub tests now model that real shape (orig wrapper + __cause__) instead of a constraint_name that no driver puts on orig directly. Tests: 76 passed. New coverage for the email-race path, both new helpers on each backend, the neutral fallback message, and the cause-chain walk. * fix(bench): match app SQLite PRAGMAs in workers; fail a sweep on any op error Two review follow-ups: - worker.py opened its SQLite engine with only connect_args timeout=30. synchronous and foreign_keys are per-connection PRAGMAs, so workers ran at SQLite's synchronous=FULL / foreign_keys=OFF while a real Gateway worker runs synchronous=NORMAL (persistence/engine.py::_enable_sqlite_wal) -- an extra fsync per commit on the measured 30%-write path, overstating SQLite's cost in the direction that flatters the "use Postgres" conclusion. Added a connect listener applying the same four PRAGMAs, with a test asserting synchronous/foreign_keys/journal_mode on a real worker connection. - summary_indicates_failure() only looked at crashes and the completed vs expected op counts, so a sweep where every op completed but raised (e.g. writes hitting OperationalError) passed as a clean measurement: completed_ops == expected, 0 crashes. Added an "errors > 0" clause; the error breakdown stays in the JSON, only the exit code changes. Test added. test_bench_concurrency.py + test_bench_worker.py green (20), plus a real 2-worker sqlite smoke run (6/6 ops, 0 errors, exit 0). * fix(auth): match the real email index name; only claim "exists" for uniqueness Review follow-ups on the classification helpers: - email is mapped_column(unique=True, index=True), which SQLAlchemy and 0001_baseline realise as a single UNIQUE INDEX (ix_users_email), not a named UNIQUE constraint. _is_email_violation compared the driver constraint name against "users_email_key", which Postgres never emits, so that arm was dead on Postgres (SQLite matched via the message). Fixed to ix_users_email. - the residual IntegrityError fallback raised "User already exists" for every remaining IntegrityError -- a NOT NULL / CHECK / foreign-key violation is not a "user already exists" condition and is not part of create_user's ValueError contract. Added _is_uniqueness_violation (sqlstate 23505, or the SQLite "UNIQUE/PRIMARY KEY constraint failed" message); only that raises the "already exists" ValueError, everything else propagates unchanged. - documented scripts/benchmark/concurrency/ in backend/AGENTS.md alongside the other benchmark family. Tests: 78 auth + 20 bench-unit pass, ruff clean. New coverage for _is_uniqueness_violation on both backends and for a non-uniqueness IntegrityError propagating out of create_user. * fix(bench): don't pre-close worker stdin (breaks communicate); require --pg-url for postgres * fix(bench): ruff format; time throughput on the op phase, not teardown - lint-backend: ruff format the files touched in this PR. - Throughput window (P2): the orchestrator sampled its wall clock after every worker's communicate() returned, so it also covered each worker's engine.dispose(), result serialization and stdout transfer. Each worker now times just its operation phase (GO -> last op) and reports ops_elapsed_s; summarize() uses max(ops_elapsed_s) over the workers -- all released by the same GO -- as the throughput window (ops_window_s). - Exercise migration 0018 (P2): test_user_oauth_partial_index.py goes through bootstrap create_all(), which builds the partial index from ORM metadata and never runs 0018.upgrade(). New Postgres-gated test_migration_0018_oauth_identity_pg_partial.py alembic-upgrades to 0017 (full index), then 0018 (asserts the predicate appears), then downgrades (asserts the full index is restored) and re-upgrades. * docs(middlewares): tighten SandboxAudit and Clarification entries in AGENTS.md PR #5134 grew agents/middlewares/AGENTS.md ~1.8 KB, pushing the effective AGENTS.md chain for that directory over the 96 KiB hard limit once this branch also documents scripts/benchmark/concurrency/ in backend/AGENTS.md. Condense the two longest middleware entries (SandboxAuditMiddleware, ClarificationMiddleware) without dropping any identifier, example, issue reference, ordering constraint, or documented gap; chain back to ~96.8 KiB. --------- Co-authored-by: Willem Jiang --- backend/AGENTS.md | 16 + .../app/gateway/auth/repositories/sqlite.py | 122 ++++++- .../deerflow/agents/middlewares/AGENTS.md | 4 +- .../deerflow/persistence/migrations/AGENTS.md | 2 + .../0018_oauth_identity_pg_partial.py | 72 ++++ .../deerflow/persistence/user/model.py | 30 +- .../concurrency/run_concurrency_bench.py | 330 ++++++++++++++++++ .../scripts/benchmark/concurrency/worker.py | 221 ++++++++++++ backend/tests/test_auth.py | 257 ++++++++++++++ backend/tests/test_bench_concurrency.py | 215 ++++++++++++ backend/tests/test_bench_worker.py | 100 ++++++ ...est_migration_0004_run_ownership_dedupe.py | 2 +- ...ration_0007_scheduled_run_active_dedupe.py | 2 +- ...t_migration_0015_scheduled_task_enqueue.py | 2 +- ...igration_0018_oauth_identity_pg_partial.py | 85 +++++ backend/tests/test_persistence_bootstrap.py | 2 +- .../test_persistence_bootstrap_concurrency.py | 2 +- .../test_persistence_bootstrap_regression.py | 4 +- .../tests/test_user_oauth_partial_index.py | 136 ++++++++ 19 files changed, 1591 insertions(+), 13 deletions(-) create mode 100644 backend/packages/harness/deerflow/persistence/migrations/versions/0018_oauth_identity_pg_partial.py create mode 100755 backend/scripts/benchmark/concurrency/run_concurrency_bench.py create mode 100644 backend/scripts/benchmark/concurrency/worker.py create mode 100644 backend/tests/test_bench_concurrency.py create mode 100644 backend/tests/test_bench_worker.py create mode 100644 backend/tests/test_migration_0018_oauth_identity_pg_partial.py create mode 100644 backend/tests/test_user_oauth_partial_index.py diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 0ad43cd76..a472bfc36 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -127,6 +127,22 @@ The offline test suite must not require network access, provider credentials, or the LongMemEval dataset. Small LongMemEval-shaped fixtures must be synthetic and generated by tests. +`scripts/benchmark/concurrency/` measures multi-process contention on the +`users` table (N separate OS processes, not asyncio tasks) for SQLite vs +Postgres -- the scenario `CONFIGURATION.md` requires Postgres for. `worker.py` +connects directly via SQLAlchemy (skipping the ~8.5s Alembic bootstrap the +orchestrator already ran once) and mirrors the app's per-connection SQLite +PRAGMAs; `run_concurrency_bench.py` seeds a disposable per-run Postgres schema, +synchronises workers on a READY/GO barrier before timing, and exits non-zero on +any crash, short op count, or `errors > 0`. Postgres runs need a throwaway +database via `--pg-url`; nothing here touches `public`. Run from `backend/`: + +```bash +uv run python scripts/benchmark/concurrency/run_concurrency_bench.py \ + --backend sqlite --workers 2,4,8,16 --ops-per-worker 50 --read-ratio 0.7 +uv run pytest tests/test_bench_concurrency.py tests/test_bench_worker.py -q +``` + ## Commands **Root directory** (for full application): diff --git a/backend/app/gateway/auth/repositories/sqlite.py b/backend/app/gateway/auth/repositories/sqlite.py index c9adeac8e..b722771c6 100644 --- a/backend/app/gateway/auth/repositories/sqlite.py +++ b/backend/app/gateway/auth/repositories/sqlite.py @@ -21,7 +21,96 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from app.gateway.auth.models import User from app.gateway.auth.repositories.base import UserNotFoundError, UserRepository -from deerflow.persistence.user.model import UserRow +from deerflow.persistence.user.model import OAUTH_IDENTITY_INDEX_NAME, UserRow + +# ``email`` is ``mapped_column(unique=True, index=True)``, which SQLAlchemy +# (and 0001_baseline) realise as a single UNIQUE INDEX -- not a named UNIQUE +# constraint -- so a Postgres duplicate reports the index name here. +_EMAIL_UNIQUE_INDEX_NAME = "ix_users_email" + + +def _driver_constraint_name(exc: IntegrityError) -> str | None: + """The violated constraint's name from the driver exception, or ``None`` + when the driver does not expose one. + + ``exc.orig`` is NOT the raw driver error. SQLAlchemy's asyncpg dialect + re-raises a plain ``AsyncAdapt_asyncpg_dbapi.IntegrityError`` built from a + rendered string and carrying only ``pgcode``/``sqlstate`` + (``sqlalchemy/dialects/postgresql/asyncpg.py::_handle_exception``); the + real ``asyncpg.UniqueViolationError`` — the one with ``constraint_name`` — + survives as ``exc.orig.__cause__`` (``raise translated_error from error``). + aiosqlite exposes no constraint name at all. Check the wrapper, then its + cause. + """ + for obj in (exc.orig, getattr(exc.orig, "__cause__", None)): + name = getattr(obj, "constraint_name", None) + if name: + return str(name) + return None + + +def _is_oauth_identity_violation(exc: IntegrityError) -> bool: + """Distinguish the ``idx_users_oauth_identity`` (oauth_provider, oauth_id) + unique-index violation from any OTHER ``IntegrityError`` reaching this + commit (a duplicate primary key, or a duplicate-email race that slipped + past the pre-check above). + + Never uses the SQLAlchemy wrapper's ``str(exc)``: it embeds the full + failed INSERT statement, whose column list names + ``oauth_provider``/``oauth_id`` on every call regardless of which + constraint fired, so a substring check on it misclassifies every + commit-time ``IntegrityError`` on this table as an OAuth conflict + (reproduced on SQLite: a duplicate ``id`` raised "OAuth account already + linked: None/None"). + + Postgres: match :data:`OAUTH_IDENTITY_INDEX_NAME` against the driver + constraint name (see :func:`_driver_constraint_name`). SQLite: no + structured name, only a message naming the columns (``"UNIQUE constraint + failed: users.oauth_provider, users.oauth_id"``) — require BOTH oauth + column names, not a bare "oauth" substring. + """ + name = _driver_constraint_name(exc) + if name is not None: + return name == OAUTH_IDENTITY_INDEX_NAME + message = str(exc.orig).lower() + return "oauth_provider" in message and "oauth_id" in message + + +def _is_email_violation(exc: IntegrityError) -> bool: + """True when ``exc`` is the ``ix_users_email`` uniqueness violation, using + the same driver-exception inspection as + :func:`_is_oauth_identity_violation` (never ``str(exc)``, whose INSERT text + names every column).""" + name = _driver_constraint_name(exc) + if name is not None: + return name == _EMAIL_UNIQUE_INDEX_NAME + return "users.email" in str(exc.orig).lower() + + +def _is_uniqueness_violation(exc: IntegrityError) -> bool: + """True for a unique-index / primary-key violation specifically, as + opposed to a NOT NULL / CHECK / foreign-key ``IntegrityError`` on the same + INSERT -- only the former means "a user like this already exists".""" + sqlstate = getattr(exc.orig, "sqlstate", None) or getattr(exc.orig, "pgcode", None) + if sqlstate is not None: + return sqlstate == "23505" # unique_violation + message = str(exc.orig).lower() + return "unique constraint failed" in message or "primary key constraint failed" in message + + +def _violated_constraint(exc: IntegrityError) -> str | None: + """Best-effort name of the constraint behind ``exc``, for a diagnostic that + does not pin an unattributed violation on a specific column. Uses the + driver constraint name where available, else the column(s) SQLite names in + its message (``"UNIQUE constraint failed: users.id"``).""" + name = _driver_constraint_name(exc) + if name: + return name + marker = "constraint failed: " + message = str(exc.orig) + if marker in message: + return message.split(marker, 1)[1].splitlines()[0].strip() or None + return None def _normalize_email(email: str) -> str: @@ -84,7 +173,11 @@ class SQLiteUserRepository(UserRepository): # ── CRUD ────────────────────────────────────────────────────────── async def create_user(self, user: User) -> User: - """Insert a new user. Raises ``ValueError`` on duplicate email. + """Insert a new user. Raises ``ValueError`` on any uniqueness + violation -- duplicate email, a duplicate (provider, oauth_id) pair + for an OAuth-linked account, or a duplicate id -- with a message + naming the specific conflict. Other ``IntegrityError``\\ s (NOT NULL, + CHECK, foreign key) propagate unchanged. The email is canonicalised to lowercase before insert so the existing unique constraint enforces case-insensitive uniqueness for new rows and @@ -103,7 +196,30 @@ class SQLiteUserRepository(UserRepository): await session.commit() except IntegrityError as exc: await session.rollback() - raise ValueError(f"Email already registered: {user.email}") from exc + # The email pre-check above already ruled out an email + # collision under normal (non-racing) conditions, so + # IntegrityErrors reaching here are usually + # idx_users_oauth_identity -- but not always (a duplicate + # primary key, or an email collision that raced past the + # pre-check). Attribute the failure to the constraint that + # actually fired instead of assuming any one of them. + if _is_oauth_identity_violation(exc): + raise ValueError(f"OAuth account already linked: {user.oauth_provider}/{user.oauth_id}") from exc + if _is_email_violation(exc): + # A duplicate address that got past the pre-check: a + # concurrent insert of the same email. + raise ValueError(f"Email already registered: {user.email}") from exc + if _is_uniqueness_violation(exc): + # Some other unique index / primary key (in practice a + # duplicate id). "Already exists" fits, but don't dress it + # up as an email conflict for an address that isn't + # registered. + constraint = _violated_constraint(exc) + raise ValueError(f"User already exists (constraint: {constraint})" if constraint else "User already exists") from exc + # A NOT NULL / CHECK / foreign-key IntegrityError is not a + # "user already exists" condition and not part of this + # method's ValueError contract -- let it propagate. + raise return user async def get_user_by_id(self, user_id: str) -> User | None: diff --git a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md index a5c1b13c0..9da25304d 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md +++ b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md @@ -64,7 +64,7 @@ it to that middleware's declaration in the same change. `__authorization_outcome` key (so `build_run_config` strips caller-supplied forgeries). Consumers pop it; the publisher and the consumer share only that contract module. -10. **SandboxAuditMiddleware** - Audits sandboxed shell/file operations for security logging before tool execution. Command classification is **defense-in-depth and audit, not a security boundary** — the sandbox itself is the isolation boundary. Command substitution is judged by *position*, not by the presence of `$(`: a substitution in **command position** (`$(curl url)`, `` `curl url` ``, the word after a `|`/`&&`/`;`, or any `eval`/`source` argument) executes fetched or interpreted content and is blocked, while **value position** (`x=$(curl url)`, `echo $(curl url)`, an argument, a `for` word list) only captures output and passes (#4611). `_HIGH_RISK_COMMAND_POSITION_PATTERNS` is therefore matched anchored against each split sub-command, never against the whole compound string, and `_split_compound_command(split_pipes=True)` supplies those sub-commands; rules that span a pipe (`| sh`, `base64 -d | ...`) still rely on `_classify_command`'s whole-command Pass 1. `_COMMAND_POSITION_PREFIX` extends the anchor over leading variable assignments and exec wrappers (`FOO=1 $(curl url)`, `env`/`command`/`builtin`/`exec`/`nohup`/`time`/`sudo`/`doas`), which are still command position; its assignment branch requires whitespace before the substitution, which is exactly what keeps `x=$(curl url)` in value position. Two execution contexts are deliberately **position-blind** and matched against the whole command in Pass 1, because they execute what they receive wherever they appear (including as an argument to something else, e.g. `xargs sh -c "$(curl url)"`): an `eval`/`source` argument, and an interpreter's **code-string flag** — `-c` (shells, `python`), `-e` (`perl`/`ruby`/`node`), `-p` (`perl`/`node`), `-r` (`php`) — plus the here-string (`<<<`) that reaches the same place through stdin. All three substitution spellings (`$(cmd`, `<(cmd`, `` `cmd ``) share one `_RISKY_SUBSTITUTION` opener so a rule cannot cover one spelling and miss another. An unquoted newline splits like `;`, because it separates statements the same way: leaving it joined let `echo hi\n$(curl url)` evade the anchored rules that its `;` spelling triggers. A heredoc body is data rather than statements, so `_split_compound_command` records headers (`<