diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4b088ea..b615cb3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,80 @@ the project follows [Semantic Versioning](https://semver.org/).
 
 ## [Unreleased]
 
+### Added
+
+- `CREATE INDEX` / `CREATE UNIQUE INDEX` on EXISTING tables now render
+  `CONCURRENTLY` by default in `plan` / `push` / `revision` output,
+  including `revision`-generated migration SQL (API: `concurrently=True`
+  on `plan` / `push` / `revision`; CLI opt-out `--no-concurrently` on
+  `push` / `revision`): the plain form takes a SHARE lock that blocks
+  all writes for the whole build, while CONCURRENTLY builds without
+  blocking writers. Indexes on tables created in the same plan stay
+  plain — a brand-new table has no concurrent writers, and the create
+  stays inside the plan's atomic transaction. Concurrently-rendered ops
+  run on push's autocommit segment (a failure there is a recorded
+  partial failure, not a rollback) and are still classified `risky`.
+  The knob threads through the advisory-lock winner path: the lock
+  winner's re-plan renders exactly what the caller requested.
+- `push` can bound each statement's runtime with a `statement_timeout`
+  (API: `push(..., statement_timeout=...)`, seconds, `None` = set
+  nothing; CLI: `push --statement-timeout`): the transactional segment
+  applies it as `SET LOCAL statement_timeout`, the concurrent
+  (autocommit) segment as a session `statement_timeout` that is `RESET`
+  before the connection returns to the pool — a pooled borrower never
+  inherits the budget. It threads through the advisory-lock winner path
+  with `concurrently`, and negative values are rejected up front with
+  a typed `SqlpushError`.
+- The concurrent (autocommit) segment of `push` now runs under a
+  session `lock_timeout` with the same value as the transactional
+  segment's: previously `CREATE INDEX CONCURRENTLY` had no lock budget
+  at all and could queue indefinitely behind another transaction's
+  table lock. The session GUC is `RESET` (and the connection
+  invalidated if the reset fails) so it never leaks to the pool's next
+  borrower.
+- Every operation in the versioned plan JSON (`diff --json`,
+  `check --json`, `Plan.to_json_dict`) now carries a `"concurrent"`
+  boolean — additive to the v1 contract, no existing key or value
+  changed. It reports whether the operation's SQL was rendered with
+  `CREATE INDEX CONCURRENTLY` (see the rendering change above).
+- `migrate` now replays CONCURRENTLY-containing chain files per-op on
+  the op-label delimiters: the plain segment runs first in one
+  transaction (chain files can create→index within one file), then
+  concurrent ops run statement-by-statement on a dedicated autocommit
+  connection (session `lock_timeout` matching the per-file txn, RESET +
+  close when the walk ends), and the versions row is written only after
+  every concurrent op succeeds — a failed concurrent op blocks the file
+  (partial failure, no versions row, strict-order stop, the already
+  committed plain segment reported honestly in the notes). Concurrent-
+  free files keep the exact 0.4.2 whole-text single-transaction replay,
+  so existing chains are byte-identically unaffected, and
+  `revision`-generated files round-trip through the chain.
+- `migrate` gains `--statement-timeout` (API:
+  `migrate(..., statement_timeout=...)`, seconds, unset by default):
+  applied as `SET LOCAL` in every per-file transaction and as a session
+  `statement_timeout` on the CONCURRENTLY autocommit connection (RESET
+  before it closes). Negative values are rejected up front with a
+  typed `SqlpushError` — same contract as `push`.
+
+### Known limitations
+
+- Generated chain files replay per-op on their `-- op N [label]`
+  delimiters; hand-edits bypass that tokenization (pinned chain spec
+  §7). A label-less body containing CONCURRENTLY routes whole to the
+  autocommit lane statement-by-statement, and lines starting `--` are
+  stripped from per-op parsing — dollar-quoted bodies containing `--`
+  lines are only safe in the whole-text fast path.
+- A failed or timed-out `CREATE INDEX CONCURRENTLY` leaves an INVALID
+  index behind: recover with `DROP INDEX CONCURRENTLY <name>` and
+  re-push (optionally `--no-concurrently`). `statement_timeout` applies
+  inside index builds, and an aborted build still leaves the INVALID
+  index.
+- Mixed-file crash window: plain segment committed + concurrent segment
+  applied + no versions row → a re-run fails loud on the existing
+  objects rather than silently re-applying.
+- Re-running against an already-existing index fails loud: the rendered
+  `CREATE INDEX CONCURRENTLY` carries no `IF NOT EXISTS`.
+
 ## [0.4.2] - 2026-09-02
 
 ### Fixed
diff --git a/README.md b/README.md
index 449f0af..9a756fa 100644
--- a/README.md
+++ b/README.md
@@ -104,8 +104,11 @@ repeated `--schema` / `--exclude` options.
 | `push` | applied | destructive blocked | error (incl. partial failure) | |
 
 `push --safe-only` runs only safe operations and skips the rest
-informationally (exit `0`). A failed `CREATE INDEX CONCURRENTLY` marks the
-run as partial failure (exit `2`) instead of silently half-applying.
+informationally (exit `0`). Indexes on existing tables build
+`CONCURRENTLY` by default (opt out with `--no-concurrently`); a failed
+`CREATE INDEX CONCURRENTLY` marks the run as partial failure (exit `2`)
+instead of silently half-applying, and leaves an INVALID index — drop it
+(`DROP INDEX CONCURRENTLY`) and re-push.
 
 ## FastAPI / SQLModel: replace `create_all`
 
@@ -140,9 +143,11 @@ flowchart LR
   internals included) before reflection even starts.
 - **Classifier** maps each operation to a risk class; unknown operations
   are `risky`, never silently safe.
-- **Executor** splits the plan: `CONCURRENTLY` statements run one-per-
-  transaction on autocommit, everything else applies in a single atomic
-  transaction with a bounded `lock_timeout`.
+- **Executor** splits the plan: existing-table indexes render
+  `CREATE INDEX CONCURRENTLY` and run one-per-transaction on autocommit
+  (`--no-concurrently` opts out; indexes on tables the same plan creates
+  stay in the atomic transaction), everything else applies in a single
+  atomic transaction with a bounded `lock_timeout`.
 - **Typed errors**: only `SqlpushError` / `ConnectFailed` /
   `MetadataImportError` escape the API, never raw driver exceptions.
 
@@ -172,12 +177,11 @@ deprecated)? There is a [migration guide](docs/migrating-from-migra.md).
 - The advisory-lock key derives from the database OID: two DSN spellings
   of the same database contend for the same lock.
 - `--json` output is a versioned contract (`"version": 1`) meant for
-  tooling; additive changes only within a version.
+  tooling; additive changes only within a version (operations now carry
+  a `concurrent` boolean).
 
 ## Roadmap (0.1.x)
 
-- `CREATE INDEX CONCURRENTLY` by default for indexes on existing tables
-- asyncpg DSN translation in `ensure_schema(AsyncEngine)`
 - jsonschema-validated `--json` output
 
 ## License
diff --git a/src/sqlpush/api.py b/src/sqlpush/api.py
index 61c0b98..c2afa69 100644
--- a/src/sqlpush/api.py
+++ b/src/sqlpush/api.py
@@ -48,8 +48,10 @@ def _raise_typed(exc: SQLAlchemyError) -> NoReturn:
     raise SqlpushError(f"database error: {exc}") from exc
 
 
-def _build_plan(metadata: MetaData, engine: Engine, schemas, exclude) -> Plan:
-    p = _engine.plan(metadata, engine, schemas=schemas, exclude=exclude)
+def _build_plan(
+    metadata: MetaData, engine: Engine, schemas, exclude, concurrently: bool = True
+) -> Plan:
+    p = _engine.plan(metadata, engine, schemas=schemas, exclude=exclude, concurrently=concurrently)
     return Plan(operations=p.operations + tuple(hypertable_operations(metadata, engine)))
 
 
@@ -65,13 +67,13 @@ class _PlannerWithDirectives(DiffEngine):
     def __init__(self, engine: DiffEngine) -> None:
         self._engine = engine
 
-    def plan(self, metadata, engine, *, schemas=None, exclude=()) -> Plan:
-        return _build_plan(metadata, engine, schemas, exclude)
+    def plan(self, metadata, engine, *, schemas=None, exclude=(), concurrently=True) -> Plan:
+        return _build_plan(metadata, engine, schemas, exclude, concurrently)
 
 
-def plan(metadata, engine, *, schemas=None, exclude=()) -> Plan:
+def plan(metadata, engine, *, schemas=None, exclude=(), concurrently: bool = True) -> Plan:
     try:
-        return _build_plan(metadata, engine, schemas, exclude)
+        return _build_plan(metadata, engine, schemas, exclude, concurrently)
     except SQLAlchemyError as exc:
         _raise_typed(exc)
 
@@ -84,9 +86,11 @@ def push(
     allow_destructive=False,
     lock=True,
     lock_timeout=5.0,
+    statement_timeout=None,
     advisory_wait=30.0,
     schemas=None,
     exclude=(),
+    concurrently=True,
 ) -> Report:
     try:
         if lock:
@@ -100,14 +104,21 @@ def push(
                 reverify=_PlannerWithDirectives(_engine),
                 schemas=schemas,
                 exclude=exclude,
+                # THREADING (pinned): the lock winner re-plans via
+                # reverify.plan(...) — both knobs must reach it and the
+                # final apply_plan, or the winner renders differently
+                # than requested
+                concurrently=concurrently,
+                statement_timeout=statement_timeout,
             )
-        p = _build_plan(metadata, engine, schemas, exclude)
+        p = _build_plan(metadata, engine, schemas, exclude, concurrently)
         return apply_plan(
             engine,
             p,
             allow_destructive=allow_destructive,
             safe_only=safe_only,
             lock_timeout=lock_timeout,
+            statement_timeout=statement_timeout,
         )
     except SQLAlchemyError as exc:
         _raise_typed(exc)
@@ -126,14 +137,23 @@ def check(metadata, engine, *, schemas=None, exclude=()) -> CheckResult:
 
 
 def revision(
-    metadata, ref_engine, *, out_dir="migrations/versions", message=None, schemas=None, exclude=()
+    metadata,
+    ref_engine,
+    *,
+    out_dir="migrations/versions",
+    message=None,
+    schemas=None,
+    exclude=(),
+    concurrently=True,
 ) -> Path:
     """Generate the next annotated-SQL migration file from models-vs-ref drift.
 
     The reference DB must sit at the chain head (caller-provided — sqlpush
     stays docker-free). Empty drift refuses loudly: no empty files.
+    ``concurrently`` follows :func:`plan` (default True: existing-table
+    indexes render CONCURRENTLY in the generated file).
     """
-    p = plan(metadata, ref_engine, schemas=schemas, exclude=exclude)
+    p = plan(metadata, ref_engine, schemas=schemas, exclude=exclude, concurrently=concurrently)
     if not p.operations:
         raise SqlpushError("no drift between models and reference DB — nothing to revise")
     risk = max((op.risk for op in p.operations), key=lambda r: RISK_RANK[r])
@@ -175,6 +195,7 @@ def migrate(
     allow_destructive=False,
     advisory_wait=30.0,
     lock_timeout=5.0,
+    statement_timeout=None,
 ) -> MigrateReport:
     """Replay annotated-SQL chain files with gates + same-txn bookkeeping.
 
@@ -182,7 +203,10 @@ def migrate(
     via ``_sync_engine_from``; engines created here are disposed).
     ``advisory_wait`` bounds the advisory-lock wait and ``lock_timeout``
     bounds each per-file transaction's lock wait (seconds; 0 = fail
-    immediately — same contract as ``push``). See
+    immediately — same contract as ``push``). ``statement_timeout``
+    (seconds, ``None`` = set nothing) bounds each statement's runtime —
+    ``SET LOCAL`` in per-file transactions, session-level on the
+    CONCURRENTLY autocommit connection. See
     ``chain.migrate.run_migrate`` for the execution contract.
     """
     engine, dispose = _sync_engine_from(target)
@@ -193,6 +217,7 @@ def migrate(
             allow_destructive=allow_destructive,
             advisory_wait=advisory_wait,
             lock_timeout=lock_timeout,
+            statement_timeout=statement_timeout,
         )
     except SQLAlchemyError as exc:
         # MigrationFileError/SqlpushError (typed) pass through untouched
diff --git a/src/sqlpush/apply/executor.py b/src/sqlpush/apply/executor.py
index 1e5af32..f1f2a1c 100644
--- a/src/sqlpush/apply/executor.py
+++ b/src/sqlpush/apply/executor.py
@@ -3,6 +3,7 @@
 
 import contextlib
 import time
+from collections.abc import Iterator
 from typing import TYPE_CHECKING
 
 from sqlalchemy import text
@@ -26,7 +27,37 @@
 
 
 def _is_concurrent(op: PlannedOperation) -> bool:
-    return "CONCURRENTLY" in op.sql.upper()
+    # Union (A5): the `concurrent` flag is authoritative for generated
+    # plans (set by the diff's injection); the SQL substring keeps
+    # hand-built Plans splitting to the autocommit segment.
+    return op.concurrent or "CONCURRENTLY" in op.sql.upper()
+
+
+@contextlib.contextmanager
+def _session_gucs(conn, **gucs: str | int) -> Iterator[None]:
+    """Session-level ``SET``/``RESET`` with pooled-connection hygiene.
+
+    The autocommit (CONCURRENTLY) segment cannot use ``SET LOCAL`` —
+    there is no surrounding transaction — so its GUCs are session-level,
+    and a session ``SET`` survives segment end (0.4.2 lesson). RESET runs
+    per GUC in ``finally``; a reset failure is suppressed (it must never
+    mask the segment's own outcome) and the connection is invalidated so
+    the pool discards it instead of handing the next borrower a session
+    still carrying the shrunken budget.
+    """
+    try:
+        for name, value in gucs.items():
+            # names come from sqlpush code, values are ints inlined in
+            # the same style as the txn segment's SET LOCAL (PostgreSQL
+            # does not accept bind parameters for SET)
+            conn.execute(text(f"SET {name} = '{value}'"))
+        yield
+    finally:
+        for name in gucs:
+            try:
+                conn.execute(text(f"RESET {name}"))
+            except Exception:  # noqa: BLE001  # never mask the segment's outcome
+                conn.invalidate()
 
 
 def apply_plan(
@@ -36,8 +67,11 @@ def apply_plan(
     allow_destructive: bool = False,
     safe_only: bool = False,
     lock_timeout: float = 5.0,
+    statement_timeout: float | None = None,
 ) -> Report:
     start = time.monotonic()
+    if statement_timeout is not None and statement_timeout < 0:
+        raise SqlpushError(f"statement_timeout must be >= 0, got {statement_timeout}")
 
     blocked = tuple(op for op in plan.operations if op.risk is RiskClass.DESTRUCTIVE)
     if blocked and not allow_destructive:
@@ -69,13 +103,21 @@ def apply_plan(
     concurrent = [op for op in runnable if _is_concurrent(op)]
     if concurrent:
         with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as conn:
-            for op in concurrent:
-                try:
-                    conn.execute(text(op.sql))
-                    applied.append(AppliedOperation(op.type, "applied"))
-                except Exception:  # noqa: BLE001  # concurrent ops record any DB failure and continue
-                    applied.append(AppliedOperation(op.type, "failed"))
-                    partial_failure = True
+            # A7: the concurrent segment gets the SAME lock budget as the
+            # txn segment (session-level: autocommit has no surrounding
+            # transaction for SET LOCAL), plus the optional statement
+            # budget; both RESET before the connection returns to the pool
+            session_gucs: dict[str, int] = {"lock_timeout": int(lock_timeout * 1000)}
+            if statement_timeout is not None:
+                session_gucs["statement_timeout"] = int(statement_timeout * 1000)
+            with _session_gucs(conn, **session_gucs):
+                for op in concurrent:
+                    try:
+                        conn.execute(text(op.sql))
+                        applied.append(AppliedOperation(op.type, "applied"))
+                    except Exception:  # noqa: BLE001  # concurrent ops record any DB failure and continue
+                        applied.append(AppliedOperation(op.type, "failed"))
+                        partial_failure = True
 
     # --- transactional segment: atomic ----------------------------------
     plain = [op for op in runnable if not _is_concurrent(op)]
@@ -83,9 +125,13 @@ def apply_plan(
         try:
             with engine.begin() as conn:
                 # NOTE: PostgreSQL does not accept bind parameters for SET
-                # (utility statement), so the int is inlined; lock_timeout is
-                # a typed float parameter, not user input.
+                # (utility statement), so the int is inlined; both timeouts
+                # are typed float parameters, not user input.
                 conn.execute(text(f"SET LOCAL lock_timeout = {int(lock_timeout * 1000)}"))
+                if statement_timeout is not None:
+                    conn.execute(
+                        text(f"SET LOCAL statement_timeout = {int(statement_timeout * 1000)}")
+                    )
                 for op in plain:
                     conn.execute(text(op.sql))
             applied.extend(AppliedOperation(op.type, "applied") for op in plain)
@@ -146,6 +192,8 @@ def with_advisory_lock(
     reverify: DiffEngine | None = None,
     schemas: Sequence[str] | None = None,
     exclude: Sequence[str] = (),
+    concurrently: bool = True,
+    statement_timeout: float | None = None,
 ) -> Report:
     """Winner migrates; losers block (bounded), then re-verify.
 
@@ -153,7 +201,9 @@ def with_advisory_lock(
     ``time.monotonic()`` deadline; once the lock is acquired the winner
     path re-plans (covering the case where the previous winner died
     mid-push). Raises :class:`SqlpushError` if the wait budget is
-    exhausted.
+    exhausted. ``concurrently`` and ``statement_timeout`` thread into
+    BOTH the winner's re-plan and its apply — the re-plan must render
+    exactly what the caller asked for.
     """
     if reverify is None:
         raise SqlpushError(
@@ -179,7 +229,13 @@ def with_advisory_lock(
         # (killed mid-push = lock silently released)
         conn.rollback()
         try:
-            plan = reverify.plan(metadata, engine, schemas=schemas, exclude=exclude)
+            plan = reverify.plan(
+                metadata,
+                engine,
+                schemas=schemas,
+                exclude=exclude,
+                concurrently=concurrently,
+            )
             if not plan.drift:
                 return Report()
             return apply_plan(
@@ -188,6 +244,7 @@ def with_advisory_lock(
                 allow_destructive=allow_destructive,
                 safe_only=safe_only,
                 lock_timeout=timeout,
+                statement_timeout=statement_timeout,
             )
         finally:
             # best-effort unlock: a secondary failure here (e.g. the
diff --git a/src/sqlpush/chain/migrate.py b/src/sqlpush/chain/migrate.py
index 1c93c75..9353e9b 100644
--- a/src/sqlpush/chain/migrate.py
+++ b/src/sqlpush/chain/migrate.py
@@ -1,13 +1,26 @@
 """migrate/stamp: chain replay + bootstrap with gates and bookkeeping.
 
-migrate execution contract (spec 2026-09-02 §5): each file's WHOLE TEXT is
-replayed in ONE ``exec_driver_sql()`` call inside a per-file transaction —
-the parsed ``ops`` are for gating/display only, NEVER reconstructed for
-execution (psycopg3 accepts multi-statement strings; nothing is tokenized,
-so dollar-quoted bodies are safe). The checksum row is inserted INSIDE the
-same transaction as the file's SQL — bookkeeping in a separate txn would
-let a crash between apply and registry re-apply the file forever
-(crash-loop over existing objects).
+migrate execution contract (spec 2026-09-02 §5, 0.5.0 hybrid): a file
+whose text contains no CONCURRENTLY replays its WHOLE TEXT in ONE
+``exec_driver_sql()`` call inside a per-file transaction — nothing is
+tokenized, so dollar-quoted bodies (and their internal ``--`` lines)
+are safe; that is the 0.4.2 fast path, byte-identical. A file that
+DOES contain CONCURRENTLY replays per-op on the op-label delimiters
+(``mf.ops``): the plain segment runs FIRST in one transaction (chain
+files can create→index within one file — a table made by a plain op
+must exist before a concurrent index on it; push's concurrent-first
+order does NOT apply here), then concurrent ops run per-op on a
+dedicated autocommit connection, and only after all of them succeed
+is the versions row written. Known cost (chain spec §7): hand-edits
+bypass the label mechanism — a label-less body containing
+CONCURRENTLY routes the whole body to the autocommit lane (executed
+statement-by-statement: the server runs a multi-statement string as
+one implicit transaction, which CONCURRENTLY refuses), and lines
+starting ``--`` are stripped from per-op parsing (they exist inside
+dollar-quoted bodies at the author's risk). Concurrent-free files
+are immune: they take the raw fast path. The crash window (plain
+committed, concurrent applied, no row yet) re-runs loud on existing
+objects — the documented chain-side cost of per-op replay.
 
 Fail-loud ordering: any blocked file (parse error, checksum mismatch,
 destructive gate, SQL failure) stops the chain — nothing later runs (R4).
@@ -19,6 +32,7 @@
 from __future__ import annotations
 
 import contextlib
+import re
 import time
 from collections.abc import Iterator
 from pathlib import Path
@@ -26,7 +40,7 @@
 from sqlalchemy import text
 from sqlalchemy.engine import Connection, Engine
 
-from sqlpush.apply.executor import advisory_key
+from sqlpush.apply.executor import _session_gucs, advisory_key
 from sqlpush.chain.format import MigrationFileError, checksum, parse_migration_file
 from sqlpush.types import MigrateReport, RiskClass, SqlpushError
 
@@ -46,6 +60,101 @@ def _chain_files(chain_dir: str | Path) -> list[Path]:
     return sorted(chain_path.glob("*.sql"))
 
 
+def _set_local_gucs(
+    conn: Connection, *, lock_timeout: float, statement_timeout: float | None
+) -> None:
+    # txn-scoped: SET LOCAL dies with the surrounding transaction (style:
+    # push's transactional segment, executor.py). NOTE: PostgreSQL does
+    # not accept bind parameters for SET (utility statement), so ints
+    # are inlined; both timeouts are typed float parameters, not user
+    # input. A chain file blocked behind another transaction's lock
+    # fails fast instead of queuing.
+    conn.execute(text(f"SET LOCAL lock_timeout = {int(lock_timeout * 1000)}"))
+    if statement_timeout is not None:
+        conn.execute(text(f"SET LOCAL statement_timeout = {int(statement_timeout * 1000)}"))
+
+
+def _record_version(conn: Connection, name: str, sha: str) -> None:
+    conn.execute(
+        text("INSERT INTO public.sqlpush_versions (name, sha256) VALUES (:n, :s)"),
+        {"n": name, "s": sha},
+    )
+
+
+_DOLLAR_TAG_RE = re.compile(r"\$[A-Za-z_][A-Za-z0-9_]*\$|\$\$")
+
+
+def _split_statements(sql: str) -> list[str]:
+    """Split a concurrent-lane op into single statements.
+
+    Physical constraint (live-verified): the server executes a
+    multi-statement simple-protocol string inside ONE implicit
+    transaction block, even on an autocommit connection — so
+    ``CREATE INDEX CONCURRENTLY`` cannot ride a multi-statement
+    ``exec_driver_sql`` call; the autocommit lane needs one statement
+    per execute. Split on top-level ``;`` only, treating single-quoted
+    strings (``''`` doubling), double-quoted identifiers, dollar-quoted
+    bodies (``$tag$...$tag$``) and ``--`` / ``/* */`` comments as
+    opaque (psql-equivalent boundaries). Generated ops are single
+    statements — the splitter is a no-op for them; only hand-edited
+    label-less bodies exercise it (chain spec §7: at the author's
+    risk). The plain segment and the fast path NEVER split.
+    """
+    statements: list[str] = []
+    buf: list[str] = []
+    i, n = 0, len(sql)
+    while i < n:
+        ch = sql[i]
+        nxt = sql[i + 1] if i + 1 < n else ""
+        if ch == "-" and nxt == "-":  # line comment
+            j = sql.find("\n", i)
+            j = n if j == -1 else j + 1
+            buf.append(sql[i:j])
+            i = j
+        elif ch == "/" and nxt == "*":  # block comment
+            j = sql.find("*/", i + 2)
+            j = n if j == -1 else j + 2
+            buf.append(sql[i:j])
+            i = j
+        elif ch == "'":  # string literal, '' doubling
+            j = i + 1
+            while j < n:
+                if sql[j] == "'" and not (j + 1 < n and sql[j + 1] == "'"):
+                    break
+                j += 2 if sql[j] == "'" else 1
+            j = min(j + 1, n)
+            buf.append(sql[i:j])
+            i = j
+        elif ch == '"':  # quoted identifier
+            j = sql.find('"', i + 1)
+            j = n if j == -1 else j + 1
+            buf.append(sql[i:j])
+            i = j
+        elif ch == "$":
+            m = _DOLLAR_TAG_RE.match(sql, i)
+            if m:  # dollar-quoted body: opaque through the closing tag
+                end = sql.find(m.group(0), m.end())
+                j = n if end == -1 else end + len(m.group(0))
+                buf.append(sql[i:j])
+                i = j
+            else:
+                buf.append(ch)
+                i += 1
+        elif ch == ";":
+            stmt = "".join(buf).strip()
+            if stmt:
+                statements.append(stmt)
+            buf = []
+            i += 1
+        else:
+            buf.append(ch)
+            i += 1
+    tail = "".join(buf).strip()
+    if tail:
+        statements.append(tail)
+    return statements
+
+
 @contextlib.contextmanager
 def _chain_session(engine: Engine, *, advisory_wait: float = 30.0) -> Iterator[Connection]:
     """Session-scoped advisory lock + versions table, shared by every verb.
@@ -97,12 +206,15 @@ def run_migrate(
     allow_destructive: bool,
     advisory_wait: float = 30.0,
     lock_timeout: float = 5.0,
+    statement_timeout: float | None = None,
 ) -> MigrateReport:
     if lock_timeout < 0:
         # same contract as push (executor.with_advisory_lock): budgets
         # are typed floats, never user input, and a negative one must
         # fail before any file or connection work
         raise SqlpushError(f"lock_timeout must be >= 0, got {lock_timeout}")
+    if statement_timeout is not None and statement_timeout < 0:
+        raise SqlpushError(f"statement_timeout must be >= 0, got {statement_timeout}")
     applied: list[str] = []
     skipped: list[str] = []
     blocked: list[str] = []
@@ -115,49 +227,115 @@ def run_migrate(
             for row in conn.execute(text("SELECT name, sha256 FROM public.sqlpush_versions"))
         }
         conn.commit()
-        for f in chain:
-            raw = f.read_text()
-            try:
-                mf = parse_migration_file(raw, name=f.name)
-            except MigrationFileError as exc:
-                blocked.append(f.name)
-                notes.append(f"{f.name}: {exc}")
-                break  # orden estricto: nada posterior corre
-            if f.name in recorded:
-                if recorded[f.name] != checksum(raw):
+        # B11: ONE lazily-created autocommit connection for the whole
+        # walk, GUC-pinned for its entire lifetime. ExitStack ordering:
+        # _session_gucs' finally (per-GUC RESET, suppress + invalidate
+        # on reset failure) unwinds BEFORE the connection close (LIFO).
+        concurrent_stack = contextlib.ExitStack()
+        concurrent_conn: Connection | None = None
+        try:
+            for f in chain:
+                raw = f.read_text()
+                try:
+                    mf = parse_migration_file(raw, name=f.name)
+                except MigrationFileError as exc:
+                    blocked.append(f.name)
+                    notes.append(f"{f.name}: {exc}")
+                    break  # orden estricto: nada posterior corre
+                if f.name in recorded:
+                    if recorded[f.name] != checksum(raw):
+                        blocked.append(f.name)
+                        notes.append(f"{f.name}: checksum mismatch (edited after apply?)")
+                        break
+                    skipped.append(f.name)
+                    continue
+                if mf.risk is RiskClass.DESTRUCTIVE and not allow_destructive:
                     blocked.append(f.name)
-                    notes.append(f"{f.name}: checksum mismatch (edited after apply?)")
+                    notes.append(f"{f.name}: DESTRUCTIVE requires --allow-destructive")
                     break
-                skipped.append(f.name)
-                continue
-            if mf.risk is RiskClass.DESTRUCTIVE and not allow_destructive:
-                blocked.append(f.name)
-                notes.append(f"{f.name}: DESTRUCTIVE requires --allow-destructive")
-                break
-            try:
-                with conn.begin():
-                    # txn-scoped: SET LOCAL dies with the per-file txn
-                    # (style: push's transactional segment, executor.py).
-                    # NOTE: PostgreSQL does not accept bind parameters for
-                    # SET (utility statement), so the int is inlined;
-                    # lock_timeout is a typed float parameter, not user
-                    # input. A chain file blocked behind another
-                    # transaction's lock fails fast instead of queuing.
-                    conn.execute(text(f"SET LOCAL lock_timeout = {int(lock_timeout * 1000)}"))
-                    # whole-file replay: exec_driver_sql bypasses text()'s
-                    # bind-param parsing entirely — ":casts" and ":=" in
-                    # hand-edited SQL must reach the server verbatim
-                    conn.exec_driver_sql(raw)
-                    conn.execute(
-                        text("INSERT INTO public.sqlpush_versions (name, sha256) VALUES (:n, :s)"),
-                        {"n": f.name, "s": checksum(raw)},
-                    )
-                applied.append(f.name)
-            except Exception as exc:  # noqa: BLE001 — report, no mask
-                blocked.append(f.name)
-                notes.append(f"{f.name}: {exc}")
-                partial = True
-                break
+                if "CONCURRENTLY" not in raw.upper():
+                    # fast path — byte-identical to 0.4.2: whole text,
+                    # one txn, one exec_driver_sql call (":casts" and
+                    # ":=" reach the server verbatim; dollar-quoted
+                    # bodies and their internal `--` lines are safe —
+                    # the parser's line-stripping quirks never execute
+                    # on this path). Checksum row rides the SAME txn.
+                    try:
+                        with conn.begin():
+                            _set_local_gucs(
+                                conn,
+                                lock_timeout=lock_timeout,
+                                statement_timeout=statement_timeout,
+                            )
+                            conn.exec_driver_sql(raw)
+                            _record_version(conn, f.name, checksum(raw))
+                        applied.append(f.name)
+                    except Exception as exc:  # noqa: BLE001 — report, no mask
+                        blocked.append(f.name)
+                        notes.append(f"{f.name}: {exc}")
+                        partial = True
+                        break
+                    continue
+                # mixed path — per-op replay on the op-label delimiters.
+                # PLAIN SEGMENT FIRST (create→index dependencies inside
+                # one file); concurrent ops afterwards, per-op on the
+                # dedicated autocommit connection; versions row LAST.
+                plain = [sql for _, sql in mf.ops if "CONCURRENTLY" not in sql.upper()]
+                conc = [sql for _, sql in mf.ops if "CONCURRENTLY" in sql.upper()]
+                plain_committed = False
+                try:
+                    if plain:
+                        with conn.begin():
+                            _set_local_gucs(
+                                conn,
+                                lock_timeout=lock_timeout,
+                                statement_timeout=statement_timeout,
+                            )
+                            for sql in plain:
+                                conn.exec_driver_sql(sql)
+                        plain_committed = True
+                    for sql in conc:
+                        if concurrent_conn is None:
+                            concurrent_conn = engine.connect().execution_options(
+                                isolation_level="AUTOCOMMIT"
+                            )
+                            concurrent_stack.callback(concurrent_conn.close)
+                            session_gucs: dict[str, int] = {
+                                "lock_timeout": int(lock_timeout * 1000)
+                            }
+                            if statement_timeout is not None:
+                                session_gucs["statement_timeout"] = int(statement_timeout * 1000)
+                            concurrent_stack.enter_context(
+                                _session_gucs(concurrent_conn, **session_gucs)
+                            )
+                        # statement-per-execute: a multi-statement string
+                        # is one implicit server txn (CONCURRENTLY refuses
+                        # it) — generated ops are single statements, so
+                        # this split only fires for hand-edited bodies
+                        for stmt in _split_statements(sql):
+                            concurrent_conn.exec_driver_sql(stmt)
+                    # never record before the concurrent ops succeed: a
+                    # row for a file whose concurrent op then failed would
+                    # be silent divergence. The crash window this leaves
+                    # (plain committed + concurrent applied + no row)
+                    # re-runs loud on existing objects — documented cost.
+                    with conn.begin():
+                        _record_version(conn, f.name, checksum(raw))
+                    applied.append(f.name)
+                except Exception as exc:  # noqa: BLE001 — report, no mask
+                    blocked.append(f.name)
+                    notes.append(f"{f.name}: {exc}")
+                    if plain_committed:
+                        # honest report: the plain segment of THIS file is
+                        # already committed (same class as push's partial
+                        # failure) and no versions row was recorded
+                        notes.append(
+                            f"{f.name}: plain segment already committed; no versions row recorded"
+                        )
+                    partial = True
+                    break
+        finally:
+            concurrent_stack.close()
         return MigrateReport(
             applied=tuple(applied),
             skipped=tuple(skipped),
diff --git a/src/sqlpush/cli.py b/src/sqlpush/cli.py
index 7e47a25..1824dd8 100644
--- a/src/sqlpush/cli.py
+++ b/src/sqlpush/cli.py
@@ -148,6 +148,8 @@ def push(
     no_lock: bool = typer.Option(False, "--no-lock"),
     lock_timeout: float = typer.Option(5.0, "--lock-timeout"),
     advisory_wait: float = typer.Option(30.0, "--advisory-wait"),
+    no_concurrently: bool = typer.Option(False, "--no-concurrently"),
+    statement_timeout: float | None = typer.Option(None, "--statement-timeout"),
     verbose: bool = typer.Option(False, "--verbose"),
     quiet: bool = typer.Option(False, "--quiet"),
     schema: SchemaOpt = None,
@@ -165,6 +167,8 @@ def push(
                 lock=not no_lock,
                 lock_timeout=lock_timeout,
                 advisory_wait=advisory_wait,
+                concurrently=not no_concurrently,
+                statement_timeout=statement_timeout,
                 schemas=schema,
                 exclude=exclude or (),
             )
@@ -223,6 +227,7 @@ def revision(
     ref_dsn: str = typer.Option(..., "--ref-dsn"),
     message: str | None = typer.Option(None, "--message", "-m"),
     out_dir: DirOpt = Path("migrations/versions"),
+    no_concurrently: bool = typer.Option(False, "--no-concurrently"),
     schema: SchemaOpt = None,
     exclude: ExcludeOpt = None,
 ):
@@ -238,6 +243,7 @@ def revision(
             engine,
             out_dir=out_dir,
             message=message,
+            concurrently=not no_concurrently,
             schemas=schema,
             exclude=exclude or (),
         )
@@ -253,6 +259,7 @@ def migrate(
     allow_destructive: bool = typer.Option(False, "--allow-destructive"),
     advisory_wait: float = typer.Option(30.0, "--advisory-wait"),
     lock_timeout: float = typer.Option(5.0, "--lock-timeout"),
+    statement_timeout: float | None = typer.Option(None, "--statement-timeout"),
     out_dir: DirOpt = Path("migrations/versions"),
 ):
     """Replay pending migration files (gates + checksum bookkeeping)."""
@@ -264,6 +271,7 @@ def migrate(
             allow_destructive=allow_destructive,
             advisory_wait=advisory_wait,
             lock_timeout=lock_timeout,
+            statement_timeout=statement_timeout,
         )
     finally:
         engine.dispose()
diff --git a/src/sqlpush/core/classify.py b/src/sqlpush/core/classify.py
index b364b9b..f487878 100644
--- a/src/sqlpush/core/classify.py
+++ b/src/sqlpush/core/classify.py
@@ -8,13 +8,18 @@
 
 
 def classify(op_type: str) -> RiskClass:
-    """add_index renders standalone: on alembic 1.19.1 even plain
-    declared indexes of NEW tables arrive standalone
-    (CreateTableOp.from_table captures columns+constraints, not
-    indexes; only instrumentation-embedded ones ride inside the
-    add_table render, and the diff dedups those away). What survives
-    runs CREATE INDEX alone — a SHARE lock that blocks writes, hence
-    risky."""
+    """add_index stays RISKY in BOTH renderings. The plain form takes a
+    SHARE lock that blocks writes for the whole build. CONCURRENTLY
+    (0.5 default for existing-table indexes) trades that write-block
+    for a different risk profile: the build is concurrent and
+    non-transactional (no rollback — a failure leaves no index behind
+    to retry with, but an aborted build can leave an INVALID index),
+    so the op is still flagged, never silently safe. Standalone
+    rendering note: on alembic 1.19.1 even plain declared indexes of
+    NEW tables arrive standalone (CreateTableOp.from_table captures
+    columns+constraints, not indexes; only instrumentation-embedded
+    ones ride inside the add_table render, and the diff dedups those
+    away)."""
     if op_type in DESTRUCTIVE:
         return RiskClass.DESTRUCTIVE
     if op_type in SAFE:
diff --git a/src/sqlpush/core/diff.py b/src/sqlpush/core/diff.py
index 69f2bbf..a26fb84 100644
--- a/src/sqlpush/core/diff.py
+++ b/src/sqlpush/core/diff.py
@@ -2,8 +2,10 @@
 
 from __future__ import annotations
 
+import dataclasses
 import fnmatch
 import io
+import re
 from collections.abc import Sequence
 
 from alembic.autogenerate import produce_migrations
@@ -464,6 +466,43 @@ def _render_op_sql(op, engine: Engine) -> str:
     return buf.getvalue().strip().rstrip(";")
 
 
+# A1: match the CREATE [UNIQUE] INDEX head of an op's (single-statement)
+# SQL. Anchored with no MULTILINE and sub'd with count=1: only the op's
+# own leading statement is touched. The lookahead makes it idempotent —
+# a second pass over an already-injected render finds CONCURRENTLY right
+# after the consumed whitespace and declines. IF NOT EXISTS lands after
+# the inserted keyword, which is the correct PostgreSQL order.
+_CONCURRENTLY_RE = re.compile(
+    r"^(CREATE\s+(?:UNIQUE\s+)?INDEX)\s+(?!CONCURRENTLY\s)", re.IGNORECASE
+)
+
+
+def _render_concurrent(ops: list[PlannedOperation]) -> list[PlannedOperation]:
+    """Render standalone ``add_index`` ops on EXISTING tables CONCURRENTLY.
+
+    Plain ``CREATE INDEX`` takes a SHARE lock that blocks all writes on
+    the table for the whole build; ``CREATE INDEX CONCURRENTLY`` is the
+    production-safe form for tables that already carry traffic. Indexes
+    on tables CREATED in the same plan stay plain: a brand-new table has
+    no concurrent writers to protect, and CONCURRENTLY cannot run inside
+    the plan's transactional create. The new-table guard keys on bare
+    table names (``op.table``), which is conservative-correct: a false
+    "new" (cross-schema bare-name collision) yields a plain
+    ``CREATE INDEX`` — always executable — while the dangerous direction
+    (injecting on a new table) cannot happen.
+    """
+    new_tables = {op.table for op in ops if op.type == "add_table"}
+    out: list[PlannedOperation] = []
+    for op in ops:
+        if op.type == "add_index" and op.table not in new_tables:
+            sql = _CONCURRENTLY_RE.sub(r"\1 CONCURRENTLY ", op.sql, count=1)
+            if sql != op.sql:
+                out.append(dataclasses.replace(op, sql=sql, concurrent=True))
+                continue
+        out.append(op)
+    return out
+
+
 class DiffEngine:
     def plan(
         self,
@@ -472,6 +511,7 @@ def plan(
         *,
         schemas: Sequence[str] | None = None,
         exclude: Sequence[str] = (),
+        concurrently: bool = True,
     ) -> Plan:
         exclude = tuple(exclude)
         # Typed `str | None` by SQLAlchemy; a dialect without a default
@@ -516,6 +556,15 @@ def plan(
             ops.extend(self._translate(op, engine, exclude))
         ops = _dedup_embedded_indexes(ops)
         ops = _dedup_enum_types(ops)
+        # ORDERING INVARIANT: CONCURRENTLY injection must run AFTER both
+        # dedups, as the final pipeline stage. _dedup_embedded_indexes
+        # matches standalone renders against the add_table render by
+        # byte containment — injecting first would rewrite only the
+        # standalone copy, break the byte-identity, and let the pair
+        # execute twice. By construction the dedup never sees a
+        # CONCURRENTLY render.
+        if concurrently:
+            ops = _render_concurrent(ops)
         return Plan(operations=tuple(ops))
 
     def _translate(self, op, engine: Engine, exclude: tuple[str, ...]) -> list[PlannedOperation]:
diff --git a/src/sqlpush/types.py b/src/sqlpush/types.py
index d6a82cb..39deeef 100644
--- a/src/sqlpush/types.py
+++ b/src/sqlpush/types.py
@@ -19,6 +19,11 @@ class PlannedOperation:
     sql: str
     table: str | None = None
     column: str | None = None
+    # True when the op's SQL was rendered with CONCURRENTLY (the
+    # executor's authoritative routing signal for generated plans;
+    # hand-built plans keep the SQL-substring fallback). Defaults False
+    # so every pre-existing construction site is unchanged.
+    concurrent: bool = False
 
 
 @dataclass(frozen=True)
@@ -44,6 +49,7 @@ def to_json_dict(self) -> dict[str, Any]:
                     "table": op.table,
                     "column": op.column,
                     "sql": op.sql,
+                    "concurrent": op.concurrent,
                 }
                 for op in self.operations
             ],
diff --git a/tests/golden/plan_v1.json b/tests/golden/plan_v1.json
index 07d9b69..5c03394 100644
--- a/tests/golden/plan_v1.json
+++ b/tests/golden/plan_v1.json
@@ -7,7 +7,8 @@
       "risk": "safe",
       "table": "t",
       "column": "a",
-      "sql": "ALTER TABLE t ADD COLUMN a INT"
+      "sql": "ALTER TABLE t ADD COLUMN a INT",
+      "concurrent": false
     }
   ],
   "sql": "ALTER TABLE t ADD COLUMN a INT"
diff --git a/tests/test_api.py b/tests/test_api.py
index b9365cf..ab04c51 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -16,6 +16,8 @@
 import sqlpush
 from sqlpush import check, ensure_schema, plan, push
 from sqlpush.annotations import hypertable
+from sqlpush.apply.executor import apply_plan
+from sqlpush.types import Plan, PlannedOperation, RiskClass
 
 pytestmark = pytest.mark.pg
 
@@ -125,6 +127,140 @@ def test_push_declared_index_applies_once(pg_engine, md_indexed):
     assert check(md_indexed, pg_engine).clean
 
 
+@pytest.fixture()
+def md_conc_push(pg_engine):
+    # §5.1: an EXISTING table gaining a declared index — the shape the
+    # CONCURRENTLY-by-default rendering targets
+    with pg_engine.begin() as conn:
+        conn.execute(text("DROP TABLE IF EXISTS conc_push"))
+        conn.execute(text("CREATE TABLE conc_push (id INTEGER PRIMARY KEY, email VARCHAR(50))"))
+    md = MetaData()
+    Table(
+        "conc_push",
+        md,
+        Column("id", Integer, primary_key=True),
+        Column("email", String(50)),
+        Index("ix_conc_push_email", "email"),
+    )
+    yield md
+    with pg_engine.begin() as conn:
+        conn.execute(text("DROP TABLE IF EXISTS conc_push"))
+
+
+def test_push_existing_table_index_concurrent_end_to_end(pg_engine, md_conc_push):
+    # Live push through the DEFAULT locked path: the index op is planned
+    # CONCURRENTLY (flag + SQL), applied on the autocommit segment, lands
+    # VALID in pg_indexes, and the push is honest about outcomes — a
+    # failing concurrent op is a partial failure while the transactional
+    # segment still applies (mirror of test_executor.py's split test,
+    # with the injected-plan shape: flag AND SQL both set).
+    rep = push(md_conc_push, pg_engine)
+    idx_applied = [a for a in rep.applied if a.type == "add_index"]
+    assert idx_applied and all(a.status == "applied" for a in idx_applied)
+    with pg_engine.connect() as conn:
+        row = conn.execute(
+            text(
+                "SELECT x.indisvalid FROM pg_indexes i "
+                "JOIN pg_class c ON c.relname = i.indexname "
+                "JOIN pg_index x ON x.indexrelid = c.oid "
+                "WHERE i.schemaname = 'public' AND i.indexname = 'ix_conc_push_email'"
+            )
+        ).first()
+    assert row is not None and row[0] is True  # exists AND not INVALID
+    assert check(md_conc_push, pg_engine).clean
+
+    plan = Plan(
+        operations=(
+            PlannedOperation(
+                type="add_index",
+                risk=RiskClass.RISKY,
+                sql="CREATE INDEX CONCURRENTLY ix_cok ON conc_push (id)",
+                table="conc_push",
+                concurrent=True,
+            ),
+            PlannedOperation(
+                type="add_index",
+                risk=RiskClass.RISKY,
+                sql="CREATE INDEX CONCURRENTLY ix_cbad ON conc_push (nope)",
+                table="conc_push",
+                concurrent=True,
+            ),
+            PlannedOperation(
+                type="add_column",
+                risk=RiskClass.SAFE,
+                sql="ALTER TABLE conc_push ADD COLUMN a2 INT",
+                table="conc_push",
+                column="a2",
+            ),
+        )
+    )
+    report = apply_plan(pg_engine, plan)
+    assert report.partial_failure is True
+    statuses = {a.type + a.status for a in report.applied}
+    assert any("failed" in s for s in statuses)
+    with pg_engine.connect() as conn:
+        has_col = conn.execute(
+            text(
+                "SELECT 1 FROM information_schema.columns "
+                "WHERE table_name = 'conc_push' AND column_name = 'a2'"
+            )
+        ).first()
+    assert has_col is not None  # txn segment still ran
+
+
+def test_concurrently_opt_out_end_to_end(pg_engine, md_conc_push, monkeypatch):
+    # A3 + the pinned threading hazard: push with concurrently=False
+    # through the DEFAULT locked path. The lock winner re-plans via
+    # reverify.plan(...), so `concurrently` MUST reach that re-plan and
+    # `statement_timeout` the final apply_plan — otherwise the winner
+    # renders differently than requested. Spies pin the kwarg threading;
+    # the captured plan and the DB pin the plain render applied.
+    import sqlpush.api as api_mod
+    from sqlpush.apply import executor as executor_mod
+    from sqlpush.core.diff import DiffEngine
+
+    plan_kwargs: list[dict] = []
+    plans: list[Plan] = []
+
+    class SpyEngine(DiffEngine):
+        def plan(self, metadata, engine, **kw):
+            plan_kwargs.append(kw)
+            p = super().plan(metadata, engine, **kw)
+            plans.append(p)
+            return p
+
+    apply_kwargs: list[dict] = []
+    real_apply = executor_mod.apply_plan
+
+    def spy_apply(engine, plan, **kw):
+        apply_kwargs.append(kw)
+        return real_apply(engine, plan, **kw)
+
+    monkeypatch.setattr(api_mod, "_engine", SpyEngine())
+    monkeypatch.setattr(executor_mod, "apply_plan", spy_apply)
+
+    rep = push(md_conc_push, pg_engine, concurrently=False, statement_timeout=4.0)
+
+    # the lock winner's re-plan was asked for — and rendered — plain
+    assert len(plan_kwargs) == 1  # exactly the winner-path re-plan
+    assert plan_kwargs[0]["concurrently"] is False
+    idx = [op for op in plans[0].operations if op.type == "add_index"]
+    assert len(idx) == 1
+    assert idx[0].sql.startswith("CREATE INDEX ")
+    assert "CONCURRENTLY" not in idx[0].sql
+    assert idx[0].concurrent is False
+    # statement_timeout threaded through with_advisory_lock to apply_plan
+    assert apply_kwargs and apply_kwargs[0]["statement_timeout"] == 4.0
+    # applied end-to-end, plain (transactional segment), schema clean
+    assert [a.status for a in rep.applied] == ["applied"]
+    with pg_engine.connect() as conn:
+        ok = conn.execute(
+            text("SELECT 1 FROM pg_indexes WHERE indexname = 'ix_conc_push_email'")
+        ).scalar()
+    assert ok is not None
+    assert check(md_conc_push, pg_engine).clean
+
+
 @pytest.fixture()
 def md_ht_schema():
     # F3a (push fire-test): @hypertable on a NON-default-schema table.
diff --git a/tests/test_cli.py b/tests/test_cli.py
index d469205..0960a67 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -3,6 +3,7 @@
 
 import json
 import os
+from pathlib import Path
 
 import pytest
 from sqlalchemy import Column, Integer, MetaData, Table, text
@@ -275,6 +276,60 @@ def test_cli_stamp_force_on_edited_file(cli_chain, tmp_path):
     assert r2.exit_code == 0
 
 
+# --- 0.5.0 flags: --no-concurrently / --statement-timeout -------------------
+
+
+def test_push_flags_route_to_api(tmp_path, monkeypatch):
+    # flag → kwarg wiring (no live DB): --no-concurrently flips the api
+    # default, --statement-timeout carries through; without the flags
+    # the api defaults (concurrently=True, statement_timeout=None) hold
+    _unit_models(tmp_path, monkeypatch)
+    seen = {}
+
+    def fake_push(*a, **k):
+        seen.update(k)
+        return Report()
+
+    monkeypatch.setattr(sqlpush.cli.api, "push", fake_push)
+    r = runner.invoke(
+        app,
+        [
+            "push",
+            "unit_models:metadata",
+            "--no-concurrently",
+            "--statement-timeout",
+            "3.5",
+            *UNIT_DSN,
+        ],
+    )
+    assert r.exit_code == 0
+    assert seen["concurrently"] is False
+    assert seen["statement_timeout"] == 3.5
+
+    seen.clear()
+    r2 = runner.invoke(app, ["push", "unit_models:metadata", *UNIT_DSN])
+    assert r2.exit_code == 0
+    assert seen["concurrently"] is True
+    assert seen["statement_timeout"] is None
+
+
+def test_revision_no_concurrently_routes_to_api(tmp_path, monkeypatch):
+    _unit_models(tmp_path, monkeypatch)
+    seen = {}
+
+    def fake_revision(*a, **k):
+        seen.update(k)
+        return Path("0001_x.sql")
+
+    monkeypatch.setattr(sqlpush.cli.api, "revision", fake_revision)
+    r = runner.invoke(
+        app,
+        ["revision", "unit_models:metadata", "--ref-dsn", UNIT_DSN[1], "--no-concurrently"],
+    )
+    assert r.exit_code == 0
+    assert seen["concurrently"] is False
+
+
 # --- unit tests below: api.push monkeypatched, no live PostgreSQL ---
 
 
diff --git a/tests/test_contract.py b/tests/test_contract.py
index 1910ab4..aa6fad1 100644
--- a/tests/test_contract.py
+++ b/tests/test_contract.py
@@ -17,13 +17,14 @@
             "type": "array",
             "items": {
                 "type": "object",
-                "required": ["type", "risk", "table", "column", "sql"],
+                "required": ["type", "risk", "table", "column", "sql", "concurrent"],
                 "properties": {
                     "type": {"type": "string"},
                     "risk": {"enum": ["safe", "risky", "destructive"]},
                     "table": {"type": ["string", "null"]},
                     "column": {"type": ["string", "null"]},
                     "sql": {"type": "string"},
+                    "concurrent": {"type": "boolean"},
                 },
             },
         },
@@ -54,3 +55,34 @@ def test_contract_valid_and_golden():
         golden.parent.mkdir(exist_ok=True)
         golden.write_text(json.dumps(payload, indent=2) + "\n")
     assert json.loads(golden.read_text()) == payload
+
+
+def test_json_contract_v1_additive_concurrent():
+    # A5: `concurrent` is ADDITIVE to JSON v1 — every operation carries
+    # it (a boolean, never null), and the pre-existing keys keep their
+    # exact values (byte-stable against the golden).
+    payload = _sample().to_json_dict()
+    op = payload["operations"][0]
+    assert op["concurrent"] is False
+    jsonschema.validate(payload, SCHEMA)
+    golden = json.loads((Path(__file__).parent / "golden" / "plan_v1.json").read_text())
+    assert golden["operations"][0]["concurrent"] is False
+    # the five v1 keys are untouched: same keys, same values
+    for key in ("type", "risk", "table", "column", "sql"):
+        assert op[key] == golden["operations"][0][key]
+    # a CONCURRENTLY-rendered op reports true
+    from sqlpush.types import Plan as _Plan
+
+    flagged = _Plan(
+        operations=(
+            PlannedOperation(
+                type="add_index",
+                risk=RiskClass.RISKY,
+                sql="CREATE INDEX CONCURRENTLY ix ON t (c)",
+                table="t",
+                concurrent=True,
+            ),
+        )
+    ).to_json_dict()
+    assert flagged["operations"][0]["concurrent"] is True
+    jsonschema.validate(flagged, SCHEMA)
diff --git a/tests/test_diff.py b/tests/test_diff.py
index e2f095d..38c4ca5 100644
--- a/tests/test_diff.py
+++ b/tests/test_diff.py
@@ -477,6 +477,127 @@ def _instrumented_geo_index(target, parent):
         Index(f"{target.name}_geo_col_idx", target.c.geo_col)
 
 
+# --- A1/A3: CONCURRENTLY-by-default for existing-table indexes --------------
+
+
+def test_concurrently_injected_for_existing_table_index_only(clean_db):
+    # A1: an index declared on an EXISTING table (no add_table op in the
+    # plan) renders CREATE INDEX CONCURRENTLY and carries the flag; a
+    # new-table-only plan carries no CONCURRENTLY anywhere (both the
+    # embedded-dedup survivors' counterparts and plain-declared new-table
+    # standalone indexes stay plain).
+    md = MetaData()
+    Table("conc_existing", md, Column("id", Integer, primary_key=True))
+    md.create_all(clean_db)
+    try:
+        md2 = MetaData()
+        Table(
+            "conc_existing",
+            md2,
+            Column("id", Integer, primary_key=True),
+            Column("email", String(50)),
+            Index("ix_conc_existing_email", "email"),
+        )
+        plan = DiffEngine().plan(md2, clean_db)
+        idx = [op for op in plan.operations if op.type == "add_index"]
+        assert [op.table for op in idx] == ["conc_existing"]
+        assert idx[0].sql.startswith("CREATE INDEX CONCURRENTLY ")
+        assert idx[0].concurrent is True
+
+        md3 = MetaData()
+        Table(
+            "conc_new",
+            md3,
+            Column("id", Integer, primary_key=True),
+            Column("email", String(50)),
+            Index("ix_conc_new_email", "email"),
+        )
+        plan2 = DiffEngine().plan(md3, clean_db)
+        assert plan2.drift  # sanity: the new-table plan is not vacuous
+        assert all("CONCURRENTLY" not in op.sql for op in plan2.operations)
+        assert all(not op.concurrent for op in plan2.operations)
+    finally:
+        with clean_db.begin() as conn:
+            conn.execute(text("DROP TABLE IF EXISTS conc_existing"))
+
+
+def test_dedup_invariant_survives_concurrent_injection(clean_db):
+    # A1 ordering: injection runs AFTER _dedup_embedded_indexes, so the
+    # dedup's byte-containment test never sees a CONCURRENTLY render.
+    # Geo-sim instrumentation (the F1/F2 mechanism): exactly one
+    # occurrence, the embedded copy stays PLAIN inside the add_table
+    # render, the standalone copy is deduped away and therefore never
+    # injected.
+    event.listen(Table, "after_parent_attach", _instrumented_geo_index)
+    try:
+        md = MetaData()
+        Table(
+            "geodup",
+            md,
+            Column("id", Integer, primary_key=True),
+            Column("geo_col", String(30)),
+        )
+        plan = DiffEngine().plan(md, clean_db)
+    finally:
+        event.remove(Table, "after_parent_attach", _instrumented_geo_index)
+    occurrences = sum("CREATE INDEX geodup_geo_col_idx" in op.sql for op in plan.operations)
+    assert occurrences == 1
+    add_table = [op for op in plan.operations if op.type == "add_table" and op.table == "geodup"]
+    assert len(add_table) == 1
+    assert "CREATE INDEX geodup_geo_col_idx" in add_table[0].sql
+    assert "CONCURRENTLY" not in add_table[0].sql
+    assert not any(op.type == "add_index" and op.table == "geodup" for op in plan.operations)
+    assert all("CONCURRENTLY" not in op.sql for op in plan.operations)
+
+
+def test_new_table_standalone_index_stays_plain(clean_db):
+    # A1 guard: a plain-declared index on a NEW table survives the
+    # embedded-index dedup trivially (standalone-only render) — the
+    # new_tables guard must keep it PLAIN (CREATE INDEX inside the same
+    # plan that creates the table is always correct and transactional).
+    md = MetaData()
+    Table(
+        "conc_new_standalone",
+        md,
+        Column("id", Integer, primary_key=True),
+        Column("email", String(50)),
+        Index("ix_conc_new_standalone_email", "email"),
+    )
+    plan = DiffEngine().plan(md, clean_db)
+    idx = [op for op in plan.operations if op.type == "add_index"]
+    assert [op.table for op in idx] == ["conc_new_standalone"]
+    assert idx[0].sql.startswith("CREATE INDEX ")
+    assert "CONCURRENTLY" not in idx[0].sql
+    assert idx[0].concurrent is False
+
+
+def test_concurrently_opt_out(clean_db):
+    # A3: concurrently=False skips the injection entirely — same plan,
+    # plain renders, no flag. (The end-to-end and lock-path halves of
+    # the opt-out live in test_api.py: they need api threading.)
+    md = MetaData()
+    Table("conc_optout", md, Column("id", Integer, primary_key=True))
+    md.create_all(clean_db)
+    try:
+        md2 = MetaData()
+        Table(
+            "conc_optout",
+            md2,
+            Column("id", Integer, primary_key=True),
+            Column("email", String(50)),
+            Index("ix_conc_optout_email", "email"),
+        )
+        plan = DiffEngine().plan(md2, clean_db, concurrently=False)
+        idx = [op for op in plan.operations if op.type == "add_index"]
+        assert len(idx) == 1
+        assert idx[0].sql.startswith("CREATE INDEX ")
+        assert "CONCURRENTLY" not in idx[0].sql
+        assert idx[0].concurrent is False
+    finally:
+        with clean_db.begin() as conn:
+            conn.execute(text("DROP TABLE IF EXISTS conc_optout"))
+
+
 def test_instrumented_index_renders_once(clean_db):
     # The REAL duplication mechanism (fire-test F1/F2): on alembic
     # 1.19.1 CreateTableOp.from_table captures columns+constraints only,
diff --git a/tests/test_executor.py b/tests/test_executor.py
index 4ba8989..199c624 100644
--- a/tests/test_executor.py
+++ b/tests/test_executor.py
@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 import pytest
-from sqlalchemy import Column, Integer, MetaData, Table, text
+from sqlalchemy import Column, Integer, MetaData, Table, create_engine, text
 
 from sqlpush.apply.executor import apply_plan
 from sqlpush.types import Plan, PlannedOperation, RiskClass, SqlpushError
@@ -133,3 +133,153 @@ def test_concurrently_split_and_partial_failure(hero_table):
     statuses = {a.type + a.status for a in report.applied}
     assert any("failed" in s for s in statuses)
     assert "a" in _cols(hero_table)  # txn segment still ran
+
+
+def test_executor_detection_flag_union(hero_table):
+    # A5: detection is op.concurrent OR the SQL substring. The flag is
+    # authoritative for generated plans; the substring keeps hand-built
+    # Plans (no flag, pre-0.5 SQL) splitting to the autocommit segment.
+    # (a) unflagged op with CONCURRENTLY SQL: routes to the concurrent
+    # segment — executed on autocommit it SUCCEEDS (in a txn it would
+    # die with "cannot run inside a transaction block" and raise).
+    plan = Plan(
+        operations=(
+            PlannedOperation(  # no flag on purpose (default False)
+                type="add_index",
+                risk=RiskClass.RISKY,
+                sql="CREATE INDEX CONCURRENTLY ix_flag_u ON hero (id)",
+                table="hero",
+            ),
+            _col_op(sql="ALTER TABLE hero ADD COLUMN a INT"),
+        )
+    )
+    report = apply_plan(hero_table, plan)
+    assert report.partial_failure is False
+    with hero_table.connect() as conn:
+        ok = conn.execute(text("SELECT 1 FROM pg_indexes WHERE indexname = 'ix_flag_u'")).scalar()
+    assert ok is not None
+    assert "a" in _cols(hero_table)
+
+    # (b) flagged op WITHOUT CONCURRENTLY SQL: routes to the concurrent
+    # segment too — a failing op there is a recorded partial failure,
+    # whereas the txn segment would raise SqlpushError (rollback).
+    plan2 = Plan(
+        operations=(
+            PlannedOperation(
+                type="add_index",
+                risk=RiskClass.RISKY,
+                sql="CREATE INDEX ix_flag_bad ON hero (nope)",
+                table="hero",
+                concurrent=True,
+            ),
+        )
+    )
+    report2 = apply_plan(hero_table, plan2)
+    assert report2.partial_failure is True
+    assert [a.status for a in report2.applied] == ["failed"]
+
+
+def _pooled(engine):
+    # a real QueuePool with exactly ONE underlying connection: whatever
+    # session state apply_plan leaves behind is handed, not recreated,
+    # to the next borrower — the 0.4.2 pooled-GUC-leak observation
+    dsn = engine.url.render_as_string(hide_password=False)
+    return create_engine(dsn, pool_size=1, max_overflow=0)
+
+
+def test_concurrent_segment_lock_timeout_set_and_reset(pg_engine, hero_table):
+    # A7: the autocommit (CONCURRENTLY) segment now runs under a session
+    # lock_timeout — an external ACCESS EXCLUSIVE holder makes the index
+    # op fail WITHIN the budget (partial failure, no unbounded queue) —
+    # and the RESET returns the pooled connection to the server default.
+    holder = pg_engine.connect()
+    pooled = _pooled(pg_engine)
+    plan = Plan(
+        operations=(
+            PlannedOperation(
+                type="add_index",
+                risk=RiskClass.RISKY,
+                sql="CREATE INDEX CONCURRENTLY ix_lt ON hero (id)",
+                table="hero",
+                concurrent=True,
+            ),
+        )
+    )
+    try:
+        with pooled.connect() as probe:
+            default = probe.execute(text("SHOW lock_timeout")).scalar()
+        holder.execute(text("LOCK TABLE hero IN ACCESS EXCLUSIVE MODE"))
+        report = apply_plan(pooled, plan, lock_timeout=1.0)
+        assert report.partial_failure is True  # failed, did not hang
+        assert report.duration < 10.0  # bounded by the ~1s budget
+        # RESET check on the SAME pool (pool_size=1 => same session):
+        # without it this borrower inherits lock_timeout='1s'
+        with pooled.connect() as probe:
+            assert probe.execute(text("SHOW lock_timeout")).scalar() == default
+    finally:
+        holder.rollback()
+        holder.close()
+    # holder released: the same op through the same pooled engine works
+    try:
+        report2 = apply_plan(pooled, plan)
+        assert report2.partial_failure is False
+        with pg_engine.connect() as conn:
+            ok = conn.execute(text("SELECT 1 FROM pg_indexes WHERE indexname = 'ix_lt'")).scalar()
+        assert ok is not None
+    finally:
+        pooled.dispose()
+
+
+def test_statement_timeout_txn_local_and_session_reset(pg_engine):
+    # B12: a not-None statement_timeout reaches BOTH segments — SET LOCAL
+    # on the transactional one, session SET (then RESET) on the
+    # autocommit one — observable via current_setting() inside each
+    # segment's execution window; None touches no GUC anywhere.
+    with pg_engine.begin() as conn:
+        conn.execute(text("DROP TABLE IF EXISTS st_probe"))
+        conn.execute(text("CREATE TABLE st_probe (seg text, val text)"))
+    pooled = _pooled(pg_engine)
+    try:
+        with pooled.connect() as probe:
+            default_st = probe.execute(text("SHOW statement_timeout")).scalar()
+
+        def _probe(seg, concurrent=False):
+            return PlannedOperation(
+                type="raw_sql",
+                risk=RiskClass.SAFE,
+                sql=f"INSERT INTO st_probe VALUES ('{seg}', current_setting('statement_timeout'))",
+                table="st_probe",
+                concurrent=concurrent,
+            )
+
+        plan = Plan(operations=(_probe("txn"), _probe("conc", concurrent=True)))
+        report = apply_plan(pooled, plan, statement_timeout=2.5)
+        assert report.partial_failure is False
+        with pooled.connect() as conn:
+            rows = dict(conn.execute(text("SELECT seg, val FROM st_probe")).all())
+            # both segments saw the injected budget
+            assert rows == {"txn": "2500ms", "conc": "2500ms"}
+            # session RESET: the pooled borrower is back at the default
+            assert conn.execute(text("SHOW statement_timeout")).scalar() == default_st
+
+        # None → no GUC touched on either segment
+        conn = pooled.connect()
+        conn.execute(text("TRUNCATE st_probe"))
+        conn.close()
+        report2 = apply_plan(pooled, plan, statement_timeout=None)
+        assert report2.partial_failure is False
+        with pooled.connect() as conn:
+            rows2 = dict(conn.execute(text("SELECT seg, val FROM st_probe")).all())
+            assert rows2 == {"txn": default_st, "conc": default_st}
+            assert conn.execute(text("SHOW statement_timeout")).scalar() == default_st
+    finally:
+        pooled.dispose()
+        with pg_engine.begin() as conn:
+            conn.execute(text("DROP TABLE IF EXISTS st_probe"))
+
+
+def test_apply_plan_rejects_negative_statement_timeout(hero_table):
+    # B12 validation mirrors the 0.4.2 lock_timeout budget pattern:
+    # negative refuses up front, typed, before anything executes
+    with pytest.raises(SqlpushError, match=">= 0"):
+        apply_plan(hero_table, Plan(), statement_timeout=-1)
diff --git a/tests/test_migrate.py b/tests/test_migrate.py
index 567421d..3ba8d24 100644
--- a/tests/test_migrate.py
+++ b/tests/test_migrate.py
@@ -1,9 +1,9 @@
 from __future__ import annotations
 
 import pytest
-from sqlalchemy import Column, Integer, MetaData, Table, text
+from sqlalchemy import Column, Index, Integer, MetaData, String, Table, text
 
-from sqlpush.api import check, migrate
+from sqlpush.api import check, migrate, revision
 from sqlpush.apply.executor import advisory_key
 from sqlpush.chain.format import MigrationFileError
 from sqlpush.types import SqlpushError
@@ -142,3 +142,202 @@ def test_migrate_advisory_wait_bounded(migrate_db, tmp_path):
         if key is not None:
             holder.execute(text("SELECT pg_advisory_unlock(:k)"), {"k": key})
         holder.close()
+
+
+# --- 0.5.0 Lane 2: hybrid CONCURRENTLY replay (B8/B11) ---------------------
+
+_CONCURRENT_TABLES = (
+    "sqlpush_versions",
+    "revt2",
+    "mixfail_host",
+    "mixfail_new",
+    "cli_xy",
+    "cli_lblless",
+    "migdq_tbl",
+)
+
+
+@pytest.fixture()
+def concurrent_db(pg_engine):
+    def _clean() -> None:
+        with pg_engine.begin() as conn:
+            for t in _CONCURRENT_TABLES:
+                conn.execute(text(f"DROP TABLE IF EXISTS {t}"))
+            conn.execute(text("DROP FUNCTION IF EXISTS migdq()"))
+
+    _clean()
+    yield pg_engine
+    _clean()
+
+
+def _label_op(n: int, label: str, sql: str) -> str:
+    return f"-- op {n} [{label}]\n{sql.rstrip(';')};\n"
+
+
+def test_generated_file_with_concurrent_op_and_dollar_quotes_replays(concurrent_db, tmp_path):
+    # T10 (pinned closed loop): revision generates a CONCURRENTLY index
+    # file for an EXISTING table; a hand-ensured op with a dollar-quoted
+    # body containing ';' replays per-op (labels delimit, nothing is
+    # statement-split), migrate applies, the versions row lands, and a
+    # re-migrate skips — idempotent.
+    with concurrent_db.begin() as conn:
+        conn.execute(text("CREATE TABLE revt2 (id integer PRIMARY KEY, note text)"))
+    md = MetaData()
+    t = Table("revt2", md, Column("id", Integer, primary_key=True), Column("note", String))
+    Index("ix_revt2_note", t.c.note)
+    out = revision(md, concurrent_db, out_dir=tmp_path, message="add index")
+    generated = out.read_text()
+    assert "CONCURRENTLY" in generated
+    # ensure the dollar-quoted op rides the same file (hand-append)
+    with out.open("a") as fh:
+        fh.write(
+            "\n"
+            + _label_op(
+                2,
+                "SAFE] add_function migdq",
+                "CREATE OR REPLACE FUNCTION migdq() RETURNS text AS $$\n"
+                "BEGIN\n  RETURN 'a;b';\nEND;\n$$ LANGUAGE plpgsql;",
+            )
+        )
+    rep = migrate(concurrent_db, chain_dir=tmp_path)
+    assert rep.applied == (out.name,) and not rep.blocked, rep.notes
+    with concurrent_db.connect() as conn:
+        assert conn.execute(text("SELECT migdq()")).scalar() == "a;b"
+        assert (
+            conn.execute(
+                text("SELECT count(*) FROM public.sqlpush_versions WHERE name = :n"),
+                {"n": out.name},
+            ).scalar()
+            == 1
+        )
+        assert conn.execute(text("SELECT count(*) FROM revt2")).scalar() == 0  # data intact
+    rep2 = migrate(concurrent_db, chain_dir=tmp_path)
+    assert rep2.applied == () and rep2.skipped == (out.name,)
+
+
+def test_plain_file_replays_whole_text_path(concurrent_db, tmp_path):
+    # T11: NO CONCURRENTLY anywhere in the file → raw fast path. The
+    # dollar-quoted body contains a line starting with '--' (and a ';'):
+    # executing it verbatim is correct; a per-op replay would STRIP that
+    # line (parser comment rule) and break the string literal. Guards
+    # against "simplifying" to always-per-op.
+    body = (
+        "-- sqlpush: revision=0001 risk=SAFE\n"
+        "CREATE OR REPLACE FUNCTION migdq() RETURNS text AS $$\n"
+        "BEGIN\n"
+        "  RETURN 'x\n"
+        "-- marker; line';\n"
+        "END;\n"
+        "$$ LANGUAGE plpgsql;\n"
+    )
+    _write(tmp_path, "0001_plain.sql", body)
+    rep = migrate(concurrent_db, chain_dir=tmp_path)
+    assert rep.applied == ("0001_plain.sql",), rep.notes
+    with concurrent_db.connect() as conn:
+        assert conn.execute(text("SELECT migdq()")).scalar() == "x\n-- marker; line"
+
+
+def test_mixed_file_plain_before_concurrent(concurrent_db, tmp_path):
+    # T12: create→index dependency INSIDE one file: the plain segment
+    # must run first (a concurrent-first order would fail: the index
+    # targets a table that does not exist yet).
+    body = (
+        "-- sqlpush: revision=0001 risk=RISKY\n"
+        + _label_op(1, "SAFE] add_table cli_xy", "CREATE TABLE cli_xy (id integer PRIMARY KEY)")
+        + "\n"
+        + _label_op(
+            2, "RISKY] add_index ix_cli_xy", "CREATE INDEX CONCURRENTLY ix_cli_xy ON cli_xy (id)"
+        )
+    )
+    _write(tmp_path, "0001_mixed.sql", body)
+    rep = migrate(concurrent_db, chain_dir=tmp_path)
+    assert rep.applied == ("0001_mixed.sql",), rep.notes
+    with concurrent_db.connect() as conn:
+        idx = conn.execute(text("SELECT 1 FROM pg_indexes WHERE indexname = 'ix_cli_xy'")).first()
+        assert idx is not None
+
+
+def test_concurrent_op_failure_blocks_file_no_versions_row(concurrent_db, tmp_path):
+    # T13: the concurrent op fails (index name already taken by the
+    # pre-existing mixfail_host index) → file blocked, partial_failure,
+    # NO versions row (the plain segment of THIS file already committed —
+    # reported honestly in notes), and strict order: later files
+    # untouched.
+    with concurrent_db.begin() as conn:
+        conn.execute(text("CREATE TABLE mixfail_host (id integer)"))
+        conn.execute(text("CREATE INDEX ix_mixfail ON mixfail_host (id)"))
+    body = (
+        "-- sqlpush: revision=0001 risk=RISKY\n"
+        + _label_op(
+            1, "SAFE] add_table mixfail_new", "CREATE TABLE mixfail_new (id integer PRIMARY KEY)"
+        )
+        + "\n"
+        + _label_op(
+            2,
+            "RISKY] add_index ix_mixfail",
+            "CREATE INDEX CONCURRENTLY ix_mixfail ON mixfail_new (id)",
+        )
+    )
+    _write(tmp_path, "0001_fail.sql", body)
+    _write(tmp_path, "0002_later.sql", SAFE_0001)
+    rep = migrate(concurrent_db, chain_dir=tmp_path)
+    assert "0001_fail.sql" in rep.blocked
+    assert rep.partial_failure is True
+    assert "0002_later.sql" not in rep.applied  # strict order holds
+    assert any("already committed" in n for n in rep.notes)  # honest report
+    with concurrent_db.connect() as conn:
+        assert (
+            conn.execute(
+                text("SELECT count(*) FROM public.sqlpush_versions WHERE name = '0001_fail.sql'")
+            ).scalar()
+            == 0
+        )  # no row: the failure is never recorded as applied
+        # the plain segment's table exists (committed), the index does not
+        assert conn.execute(text("SELECT count(*) FROM mixfail_new")).scalar() == 0
+        assert (
+            conn.execute(
+                text("SELECT count(*) FROM pg_indexes WHERE indexname = 'ix_mixfail'")
+            ).scalar()
+            == 1
+        )  # only the pre-existing one on mixfail_host
+
+
+def test_no_if_not_exists_generated_and_rerun_fails_loud(concurrent_db, tmp_path):
+    # T14: generated concurrent SQL carries no IF NOT EXISTS guard — the
+    # documented crash-window semantics: with the versions row gone (crash
+    # after apply, before/mid bookkeeping), a re-run fails LOUD on the
+    # existing index instead of silently skipping.
+    with concurrent_db.begin() as conn:
+        conn.execute(text("CREATE TABLE revt2 (id integer PRIMARY KEY, note text)"))
+    md = MetaData()
+    t = Table("revt2", md, Column("id", Integer, primary_key=True), Column("note", String))
+    Index("ix_revt2_note", t.c.note)
+    out = revision(md, concurrent_db, out_dir=tmp_path, message="add index")
+    generated = out.read_text()
+    assert "CONCURRENTLY" in generated
+    assert "IF NOT EXISTS" not in generated
+    rep = migrate(concurrent_db, chain_dir=tmp_path)
+    assert rep.applied == (out.name,)
+    # simulate the crash window: bookkeeping lost, objects present
+    with concurrent_db.begin() as conn:
+        conn.execute(text("DELETE FROM public.sqlpush_versions"))
+    rep2 = migrate(concurrent_db, chain_dir=tmp_path)
+    assert out.name in rep2.blocked and rep2.partial_failure is True
+
+
+def test_hand_edit_concurrent_in_labelless_file_routes_autocommit(concurrent_db, tmp_path):
+    # T15: label-less body containing CREATE INDEX CONCURRENTLY — the
+    # parser yields ONE op (whole body), which routes to the autocommit
+    # lane: everything applies (documented hand-edit behavior, chain
+    # spec §7).
+    body = (
+        "-- sqlpush: revision=0001 risk=RISKY\n"
+        "CREATE TABLE cli_lblless (id integer PRIMARY KEY);\n"
+        "CREATE INDEX CONCURRENTLY ix_lblless ON cli_lblless (id);\n"
+    )
+    _write(tmp_path, "0001_lblless.sql", body)
+    rep = migrate(concurrent_db, chain_dir=tmp_path)
+    assert rep.applied == ("0001_lblless.sql",), rep.notes
+    with concurrent_db.connect() as conn:
+        idx = conn.execute(text("SELECT 1 FROM pg_indexes WHERE indexname = 'ix_lblless'")).first()
+        assert idx is not None
diff --git a/tests/test_types.py b/tests/test_types.py
index b0ecebd..1ef2f2f 100644
--- a/tests/test_types.py
+++ b/tests/test_types.py
@@ -35,8 +35,19 @@ def test_json_contract_v1_shape():
     assert d["version"] == 1
     assert d["drift"] is True
     op = d["operations"][0]
-    assert set(op) == {"type", "risk", "table", "column", "sql"}
+    assert set(op) == {"type", "risk", "table", "column", "sql", "concurrent"}
     assert op["risk"] == "safe"
+    assert op["concurrent"] is False
+
+
+def test_planned_operation_concurrent_defaults_false():
+    # additive field: every pre-existing construction site (tests,
+    # hand-built plans) keeps working unchanged; injection flips it
+    # via dataclasses.replace on exactly the ops it rewrites
+    op = PlannedOperation(
+        type="add_index", risk=RiskClass.RISKY, sql="CREATE INDEX ix ON t (c)", table="t"
+    )
+    assert op.concurrent is False
 
 
 def test_exception_hierarchy():
diff --git a/tests/test_validation.py b/tests/test_validation.py
index 87116d8..804bb8d 100644
--- a/tests/test_validation.py
+++ b/tests/test_validation.py
@@ -11,9 +11,9 @@
 from sqlalchemy.pool import NullPool
 
 from sqlpush import api
-from sqlpush.apply.executor import with_advisory_lock
+from sqlpush.apply.executor import apply_plan, with_advisory_lock
 from sqlpush.core.diff import DiffEngine
-from sqlpush.types import ConnectFailed, RiskClass, SqlpushError
+from sqlpush.types import ConnectFailed, Plan, RiskClass, SqlpushError
 
 UNREACHABLE = "postgresql+psycopg://u:p@127.0.0.1:5999/x?connect_timeout=2"
 
@@ -86,6 +86,24 @@ def test_migrate_rejects_negative_budgets(kwargs, tmp_path):
         api.migrate(_lazy_engine(), chain_dir=tmp_path, **kwargs)
 
 
+# --- B12 (0.5.0): migrate statement_timeout validation (validation runs
+# before any connection attempt, so DB-free like I4) ------------------------
+
+
+def test_statement_timeout_negative_rejected(tmp_path):
+    with pytest.raises(SqlpushError, match=">= 0"):
+        api.migrate(_lazy_engine(), chain_dir=tmp_path, statement_timeout=-1)
+
+
+# --- B12 (0.5.0): push statement_timeout budget validation (validation
+# runs before any connection attempt, so DB-free like I4/B3) ---------------
+
+
+def test_apply_plan_rejects_negative_statement_timeout():
+    with pytest.raises(SqlpushError, match=">= 0"):
+        apply_plan(_lazy_engine(), Plan(), statement_timeout=-1)
+
+
 # --- I6: AlterColumnOp disambiguation (sentinel semantics per
 # docs/notes/alembic-notes.md Pattern C: False/None = do not touch) ---------