#54: busy_timeout=30000 on hydrate writer connections (fixes parallel-worker crashes)

2026-05-27 v4-fixed bench on 3090: 3 of 4 parallel workers crashed with
sqlite3.OperationalError: database is locked at PRAGMA journal_mode =
MEMORY. Result: only shard 000 fully hydrated, shards 001/002/003 had
chunks rows landed but content NULL (~99.99% empty), no FTS data.
Peer functionally usable for only ~25% of corpus.

Root cause: parallel hydrate (xargs -P N) spawns N separate processes
that each open connections to all M target shards. Multiple processes
attempting PRAGMA journal_mode change on the same .db file at the same
instant serialize on a brief exclusive lock — without busy_timeout
SQLite throws BUSY immediately and the worker exception-propagates
out of _pull_pack_inner_routed before phase 2 (chunks fill) can run.

Fix: `PRAGMA busy_timeout = 30000` as the FIRST statement on every
hydrate writer connection. SQLite then waits up to 30s for any lock
instead of throwing — slowest worker gets its lock, fastest writes
go through immediately. No measurable cost when there's no
contention (busy_timeout is a wait, not a poll).

Set in TWO places:
  arborist/cli.py:_cmd_cold_unpack
    Immediately after connect(p), before any other PRAGMA. Covers
    the PRAGMA foreign_keys = OFF that runs before the hydrate
    pipeline.
  arborist/evict.py:_pull_pack_inner_routed
    Belt-and-suspenders: in case the function is called with
    externally-built connections that didn't set busy_timeout, the
    bulk-tuning loop sets it before journal_mode/synchronous/etc.

6 cold-unpack-routed tests pass (no contention in single-process
test fixture; the fix is invisible there). Real validation is the
next 3090 v5 bench against re-packed (#53 pre-sized chunks) bucket.

arborist/cli.py

@@ -467,6 +467,8 @@ def _cmd_ask(args: argparse.Namespace) -> int:
     call_policy = dict(_DEFAULT_ASK_POLICY)
     if getattr(args, "answer_mode", None):
         call_policy["answer_mode"] = args.answer_mode
+    if getattr(args, "user_payload_layout", None):
+        call_policy["user_payload_layout"] = args.user_payload_layout
     try:
         result = ask(
             conn,
@@ -533,6 +535,8 @@ def _cmd_query(args: argparse.Namespace) -> int:
         call_policy["question_dedup"] = args.question_dedup
     if getattr(args, "answer_mode", None):
         call_policy["answer_mode"] = args.answer_mode
+    if getattr(args, "user_payload_layout", None):
+        call_policy["user_payload_layout"] = args.user_payload_layout
     if getattr(args, "repair", False):
         # Mechanical-only repair when --repair is set; --repair-reprompts
         # adds the optional re-prompt tier on top. Both default off so
@@ -3008,6 +3012,11 @@ def _cmd_cold_unpack(args: argparse.Namespace) -> int:
         targets: list = []
         for p in target_paths:
             t = connect(p)
+            # busy_timeout FIRST so all subsequent statements wait
+            # instead of crashing with `database is locked` under
+            # parallel-hydrate (xargs -P N spawns N processes that
+            # each open connections to all M target shards). #54.
+            t.execute("PRAGMA busy_timeout = 30000")
             t.execute("PRAGMA foreign_keys = OFF")
             targets.append(t)
         try:
@@ -5257,6 +5266,14 @@ def build_parser() -> argparse.ArgumentParser:
             "pre-parser; pairs with grammar-constrained inference)."
         ),
     )
+    ask_cmd.add_argument(
+        "--user-payload-layout", dest="user_payload_layout", default=None,
+        choices=["tail", "bookend", "per_chunk"],
+        help=(
+            "where the question text appears in the final user message. "
+            "See `query --user-payload-layout` for full semantics."
+        ),
+    )
     ask_cmd.set_defaults(func=_cmd_ask)
 
     query_cmd = sub.add_parser(
@@ -5366,6 +5383,19 @@ def build_parser() -> argparse.ArgumentParser:
             "constrained inference like Qwen 3.6 reasoner / Claude / GPT-4)."
         ),
     )
+    query_cmd.add_argument(
+        "--user-payload-layout", dest="user_payload_layout", default=None,
+        choices=["tail", "bookend", "per_chunk"],
+        help=(
+            "where the question text appears in the final user message. "
+            "'tail' (default, preserves prior cache): question after "
+            "evidence only. 'bookend': question repeated before AND "
+            "after evidence — counters lost-in-the-middle on small "
+            "models (Hermes-3-8B) with long contexts. 'per_chunk': "
+            "bookend + a one-line `[for: <question>]` reminder before "
+            "each evidence block. Folds into governance_policy_hash."
+        ),
+    )
     query_cmd.add_argument(
         "--retrieval-keywords", dest="retrieval_keywords", default=None,
         help=(

arborist/evict.py

@@ -1141,7 +1141,18 @@ def _pull_pack_inner_routed(
     # not at the schema level, so they affect only this writer's
     # behaviour. Crash mid-hydrate is recoverable (re-pull from
     # bucket), so durability of intermediate state has no value.
+    #
+    # busy_timeout MUST come FIRST: in parallel-hydrate (xargs -P N),
+    # N workers in separate processes each open connections to all M
+    # target shards and apply PRAGMAs. Multiple processes attempting
+    # PRAGMA journal_mode change on the same file at the same instant
+    # serialize on a brief exclusive lock; without busy_timeout they
+    # throw `database is locked` and abort. With busy_timeout SQLite
+    # waits up to 30s and retries. 2026-05-27 v4-fixed bench: 3 of 4
+    # workers crashed at this exact PRAGMA; only shard 000 ended up
+    # fully hydrated. Fix #54.
     for t in targets:
+        t.execute("PRAGMA busy_timeout = 30000")   # 30 sec wait, not throw
         t.execute("PRAGMA synchronous = OFF")
         t.execute("PRAGMA journal_mode = MEMORY")  # no on-disk WAL
         t.execute("PRAGMA temp_store = MEMORY")