qa: providence_cache INSERT — ON CONFLICT(cache_key) DO NOTHING

The cache lookup in ask()/query()/canonical-persist runs outside the
write transaction, so two concurrent callers on the same cache_key can
both miss and both reach the providence_cache INSERT — the loser raised
`UNIQUE constraint failed: providence_cache.cache_key` and ask() crashed
(MOAD-0005 / TOCTOU; sibling of the af870bb append_audit fix). All three
write sites (qa/runner.py, qa/query.py, qa/canonical_cache.py) now end in
`ON CONFLICT(cache_key) DO NOTHING`, so the loser no-ops (its answer is
equivalent — same question/model/policy ⇒ same cache_key; canonical
answers are deterministic). With busy_timeout on every connection
(af870bb) the loser waits on the writer's lock then no-ops.

test_qa.py::test_concurrent_ask_same_cache_key_no_unique_crash — 6
threads run ask() on the same question concurrently; must not raise;
exactly one cache row lands. Verified it fails without the fix (5 of 6
threads raise IntegrityError).

arborist/qa/canonical_cache.py

@@ -248,6 +248,10 @@ def persist_canonical(
             ts=now,
         )
         qa_conn.execute(
+            # ON CONFLICT(cache_key) DO NOTHING: canonical-projection answers
+            # are deterministic, so a concurrent peer persisting the same
+            # cache_key writes an identical row — the loser no-ops instead of
+            # raising UNIQUE constraint failed.
             "INSERT INTO providence_cache "
             "(cache_key, source_root, document_uri, question_hash, question_text, "
             " answer_text, merkle_proof, model_profile_hash, conversation_hash, "
@@ -256,7 +260,8 @@ def persist_canonical(
             " created_at, hit_count, audit_mode, n_quotes, n_verified, "
             " unverified_quotes, verifier_method, run_dag_root, run_dag_blob) "
             "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'live', ?, ?, ?, 0, "
-            " ?, ?, ?, ?, ?, ?, ?)",
+            " ?, ?, ?, ?, ?, ?, ?) "
+            "ON CONFLICT(cache_key) DO NOTHING",
             (
                 cache_key_value,
                 src,

arborist/qa/query.py

@@ -3403,6 +3403,10 @@ def query(
                 ts=now,
             )
             qa_conn.execute(
+                # ON CONFLICT(cache_key) DO NOTHING: the cache lookup runs
+                # outside this transaction, so two concurrent queries on the
+                # same cache_key can both miss and both reach this INSERT — the
+                # loser no-ops instead of raising UNIQUE constraint failed.
                 "INSERT INTO providence_cache "
                 "(cache_key, source_root, document_uri, question_hash, question_text, "
                 " answer_text, merkle_proof, model_profile_hash, conversation_hash, "
@@ -3411,7 +3415,8 @@ def query(
                 " created_at, hit_count, audit_mode, n_quotes, n_verified, "
                 " unverified_quotes, verifier_method, run_dag_root, run_dag_blob) "
                 "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'live', ?, ?, ?, 0, "
-                " ?, ?, ?, ?, ?, ?, ?)",
+                " ?, ?, ?, ?, ?, ?, ?) "
+                "ON CONFLICT(cache_key) DO NOTHING",
                 (
                     ckey,
                     context_root,

arborist/qa/runner.py

@@ -1128,6 +1128,11 @@ def ask(
             ts=now,
         )
         conn.execute(
+            # ON CONFLICT(cache_key) DO NOTHING: the cache lookup above runs
+            # outside this transaction, so two concurrent ask()s on the same
+            # cache_key can both miss and both reach this INSERT — the loser
+            # no-ops instead of raising UNIQUE constraint failed (its answer is
+            # equivalent: same question/model/policy ⇒ same cache_key).
             "INSERT INTO providence_cache "
             "(cache_key, source_root, document_uri, question_hash, question_text, "
             " answer_text, merkle_proof, model_profile_hash, conversation_hash, "
@@ -1136,7 +1141,8 @@ def ask(
             " created_at, hit_count, audit_mode, n_quotes, n_verified, "
             " unverified_quotes, verifier_method, run_dag_root, run_dag_blob) "
             "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'live', ?, ?, ?, 0, "
-            " ?, ?, ?, ?, ?, ?, ?)",
+            " ?, ?, ?, ?, ?, ?, ?) "
+            "ON CONFLICT(cache_key) DO NOTHING",
             (
                 ckey,
                 document_root,

tests/test_qa.py

@@ -233,3 +233,61 @@ def test_cache_key_is_pure_function():
         "|".join([src_root, qh, mh, ch, gh, "v1", "v1", "v1"]).encode()
     ).hexdigest()
     assert k1 == expected
+
+
+def test_concurrent_ask_same_cache_key_no_unique_crash(tmp_path):
+    """Two threads run ``ask()`` on the same question/model concurrently.
+
+    The cache lookup happens *outside* the write transaction, so both miss
+    and both reach the ``providence_cache`` INSERT — ``ON CONFLICT(cache_key)
+    DO NOTHING`` makes the loser no-op instead of raising
+    ``UNIQUE constraint failed: providence_cache.cache_key``. Exactly one
+    cache row lands. (Without the fix the losers raise; with ``busy_timeout``
+    on every connection the loser waits on the writer's lock then no-ops.)
+    """
+    import threading
+
+    db = tmp_path / "race-qa.db"
+    conn = connect(db)
+    try:
+        root = _ingest_one(conn)
+    finally:
+        conn.close()
+
+    n_threads = 6
+    barrier = threading.Barrier(n_threads)
+    errors: list[BaseException] = []
+    lock = threading.Lock()
+
+    def worker() -> None:
+        try:
+            c = connect(db)
+            try:
+                barrier.wait(timeout=20)
+                ask(
+                    c,
+                    document_root=root,
+                    question="What are the forms of capital?",
+                    client=StubClient(answer="The forms include living and social."),
+                    model_id="test-model",
+                )
+            finally:
+                c.close()
+        except BaseException as exc:  # noqa: BLE001 — surface it
+            with lock:
+                errors.append(exc)
+
+    threads = [threading.Thread(target=worker) for _ in range(n_threads)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join(timeout=40)
+
+    assert not errors, f"concurrent ask() raised: {errors!r}"
+
+    conn = connect(db)
+    try:
+        n_rows = conn.execute("SELECT COUNT(*) FROM providence_cache").fetchone()[0]
+    finally:
+        conn.close()
+    assert n_rows == 1, f"expected exactly one providence_cache row, got {n_rows}"