---
phase: 65.1-phase-65-code-review-remediation
plan: 05
type: execute
wave: 1
depends_on: []
files_modified:
  - test/integration/test_concurrent_writes_per_call_conn.py
autonomous: true
requirements:
  - REM-WR-04

must_haves:
  truths:
    - "A new Python integration test exercises N (>=4) threads each running a mix of CREATE / DROP / ALTER on overlapping semantic-view names with a wall-budget bound (≤30s)"
    - "Per-thread outcomes are categorised as `success | already_exists | constraint_violation | unknown_error`"
    - "Assertion: zero `unknown_error` outcomes — any unexpected error shape fails the test loudly"
    - "Assertion: for any race set on the same view name across threads, the count of `success`-marked CREATEs aligns with the final `list_semantic_views()` content (minus `success`-marked DROPs)"
    - "No thread hangs; wall budget honored"
  artifacts:
    - path: test/integration/test_concurrent_writes_per_call_conn.py
      provides: "WR-04 concurrent-writes integration test (D-19) with PK-violation tolerance per TECH-DEBT 23 (D-20)"
      contains: "constraint_violation"
  key_links:
    - from: test/integration/test_concurrent_writes_per_call_conn.py
      to: test/integration/test_concurrent_reads_per_call_conn.py
      via: "verbatim fixture (PEP-723 + open_master + threading.Event gate)"
      pattern: "open_master"
---

<objective>
Create `test/integration/test_concurrent_writes_per_call_conn.py` mirroring the existing reads-side template (`test/integration/test_concurrent_reads_per_call_conn.py`). N threads each cycle through a mix of `CREATE SEMANTIC VIEW IF NOT EXISTS v_{0..3}`, `DROP SEMANTIC VIEW IF EXISTS v_{0..3}`, and `ALTER SEMANTIC VIEW v_{0..3} SET COMMENT '...'` targeting overlapping view names to maximise contention.

Per-call outcomes are categorised into four buckets:
- `success` — operation completed without error
- `already_exists` — caught error matching the canonical "already exists" wording (e.g. `Catalog Error: ...already exists` from the race-guard SQL)
- `constraint_violation` — PK violation per TECH-DEBT 23 (`Constraint Error: Duplicate key "name: v_X" violates primary key constraint`)
- `unknown_error` — anything else

The test asserts:
1. Zero `unknown_error` outcomes (any unexpected error shape fails loudly with full diagnostic text)
2. No thread hangs — all threads `join()` within wall budget (30s)
3. Final-state check: `list_semantic_views()` output is consistent with the union of `success`-marked CREATEs minus the union of `success`-marked DROPs

This is purely a new test file — no production source changes. Runs independently of all other plans in Wave 1. Plan 01 does NOT scaffold this file (the reads-template is sufficiently close that one task suffices).

Purpose: WR-04 finding noted Phase 65's concurrency coverage was reads-only; writes need exercising under the new parser_override + race-guard SQL design (Phase 60). The test pins both the no-hang property and the constraint-violation tolerance documented in TECH-DEBT 23.

Implements D-19, D-20 (WR-04).
</objective>

<execution_context>
@$HOME/.claude/get-shit-done/workflows/execute-plan.md
@$HOME/.claude/get-shit-done/templates/summary.md
</execution_context>

<context>
@.planning/PROJECT.md
@.planning/STATE.md
@.planning/phases/65.1-phase-65-code-review-remediation/65.1-CONTEXT.md
@.planning/phases/65.1-phase-65-code-review-remediation/65.1-RESEARCH.md
@.planning/phases/65.1-phase-65-code-review-remediation/65.1-PATTERNS.md
@test/integration/test_concurrent_reads_per_call_conn.py
@TECH-DEBT.md
</context>

<tasks>

<task type="auto">
  <name>Task 1: Create test_concurrent_writes_per_call_conn.py</name>
  <read_first>
    - test/integration/test_concurrent_reads_per_call_conn.py (FULL FILE — the verbatim template, ~150-200 LOC; preserve PEP-723 header, EXT_DIR/EXT_PATH helpers, open_master, WorkerResult shape, threading.Event start-gate, wall-budget assertion)
    - .planning/phases/65.1-phase-65-code-review-remediation/65.1-PATTERNS.md lines 640-689 (test design notes including PK-violation tolerance categorisation)
    - .planning/phases/65.1-phase-65-code-review-remediation/65.1-RESEARCH.md lines 647-666 (template confirmation + D-20 finding on TECH-DEBT 23)
    - TECH-DEBT.md lines 200-205 (the documented PK-violation race shape — exact wording the test must tolerate)
    - test/integration/test_ducklake_helpers.py (find this file under test/integration; it exports `get_ext_dir`, `get_extension_path` — used by the reads template)
  </read_first>
  <files>test/integration/test_concurrent_writes_per_call_conn.py</files>
  <action>
Create the new file. Start by literally copying the structure of `test_concurrent_reads_per_call_conn.py` then swap:

**Header / boilerplate (verbatim):**
- PEP-723 header from reads template lines 1-5
- `sys.path.insert` + `from test_ducklake_helpers import get_ext_dir, get_extension_path` block
- Constants section — bump `NUM_THREADS = 8`, `CALLS_PER_THREAD = 10`, `WALL_BUDGET_SEC = 30.0` (same as reads)
- `open_master(db_path)` function — identical
- A new constant `VIEW_NAMES = ["v_0", "v_1", "v_2", "v_3"]` (overlapping name pool — 8 threads cycling through 4 names maximises contention)

**SETUP_SQL list** (run once before threads start, on the master connection):
- `CREATE TABLE IF NOT EXISTS t1 (id INTEGER, name VARCHAR);`
- (no pre-existing semantic views — the threads race to CREATE them)

**WorkerResult dataclass:**
```python
from dataclasses import dataclass, field
@dataclass
class WorkerResult:
    success: int = 0
    already_exists: int = 0
    constraint_violation: int = 0
    unknown_errors: list = field(default_factory=list)
    create_successes: list = field(default_factory=list)  # view names
    drop_successes: list = field(default_factory=list)
    ok: bool = False
```

**categorise(exc, view_name) helper:**
```python
def categorise(exc):
    msg = str(exc)
    if "already exists" in msg:
        return "already_exists"
    if "Constraint Error" in msg and "Duplicate key" in msg and "primary key constraint" in msg:
        return "constraint_violation"
    return "unknown_error"  # caller records full text
```

**worker function:**
```python
def worker(thread_conn, thread_id, result, gate, view_names):
    import random
    rng = random.Random(thread_id)
    gate.wait()
    deadline = time.time() + WALL_BUDGET_SEC
    for i in range(CALLS_PER_THREAD):
        if time.time() > deadline:
            result.unknown_errors.append(f"thread {thread_id} exceeded wall budget mid-run")
            return
        view = rng.choice(view_names)
        op = rng.choice(["create", "drop", "alter"])
        try:
            if op == "create":
                thread_conn.execute(
                    f"CREATE SEMANTIC VIEW IF NOT EXISTS {view} "
                    f"TABLES (t1 AS t1) METRICS (t1.cnt AS count(*))"
                )
                result.success += 1
                result.create_successes.append(view)
            elif op == "drop":
                thread_conn.execute(f"DROP SEMANTIC VIEW IF EXISTS {view}")
                result.success += 1
                result.drop_successes.append(view)
            else:  # alter
                thread_conn.execute(
                    f"ALTER SEMANTIC VIEW {view} SET COMMENT 'thread_{thread_id}_iter_{i}'"
                )
                result.success += 1
        except Exception as exc:
            cat = categorise(exc)
            if cat == "already_exists":
                result.already_exists += 1
            elif cat == "constraint_violation":
                result.constraint_violation += 1
            else:
                result.unknown_errors.append(f"thread {thread_id} op={op} view={view}: {exc!r}")
    result.ok = True
```

**Main test body (`def test_concurrent_writes_per_call_conn():`):**
1. `tempfile.TemporaryDirectory()` → `db_path`
2. `master = open_master(db_path)`; run SETUP_SQL on `master`
3. `cursors = [master.cursor() for _ in range(NUM_THREADS)]`
4. `results = [WorkerResult() for _ in range(NUM_THREADS)]`
5. `gate = threading.Event()`
6. `threads = [Thread(target=worker, args=(cursors[i], i, results[i], gate, VIEW_NAMES)) for i in range(NUM_THREADS)]`
7. Start all threads; `gate.set()`; `start = time.time()`
8. Join each thread with `timeout=WALL_BUDGET_SEC + 5`; assert `not t.is_alive()` (no hang)
9. `elapsed = time.time() - start`; assert `elapsed < WALL_BUDGET_SEC + 5`
10. For each result: assert `r.ok`; assert `not r.unknown_errors` with diagnostic dump
11. **Final-state check:** Query `list_semantic_views()` on `master`; build expected set as `union(create_successes) - union(drop_successes)`. The actual final set must be a subset of expected (some CREATEs may have happened after DROPs that succeeded earlier — exact equality is too strict). Assert that every name in the actual final set was in `union(create_successes)` (so we didn't conjure rows from nowhere). Document the looser invariant in a comment.

**`if __name__ == "__main__": pytest.main([__file__, "-v"])`**

NOTE on the final-state check: per researcher Pitfall 4 and Assumption A4, race ordering means we cannot pin exact equality. The looser "no spurious rows" check + "no `unknown_error`" check is the robust contract.
  </action>
  <verify>
    <automated>uv run test/integration/test_concurrent_writes_per_call_conn.py &gt; /tmp/claude/p05_run.log 2>&amp;1; RC=$?; tail -60 /tmp/claude/p05_run.log; test $RC -eq 0</automated>
  </verify>
  <acceptance_criteria>
    - `test/integration/test_concurrent_writes_per_call_conn.py` exists with PEP-723 header
    - Imports duckdb, threading, tempfile; uses `from test_ducklake_helpers import get_ext_dir, get_extension_path`
    - Defines `worker(...)`, `categorise(...)`, `WorkerResult`, `test_concurrent_writes_per_call_conn()`
    - Worker exercises a mix of CREATE/DROP/ALTER on `VIEW_NAMES` (length 4) with `NUM_THREADS >= 4`
    - Per-call exception categorisation handles 3 known classes + records unknowns with full diagnostic text
    - Test asserts: no hangs, no `unknown_errors`, no spurious views in final state, wall budget honored
    - `uv run test/integration/test_concurrent_writes_per_call_conn.py` exits 0
    - Total wall time observed under 30 seconds
  </acceptance_criteria>
  <done>WR-04 concurrent writes coverage live; PK-violation tolerance per D-20; test green.</done>
</task>

</tasks>

<threat_model>
## Trust Boundaries

| Boundary | Description |
|----------|-------------|
| Multiple concurrent DDL writers on same view names | Race conditions between CREATE/DROP/ALTER on overlapping names |

## STRIDE Threat Register

| Threat ID | Category | Component | Disposition | Mitigation Plan |
|-----------|----------|-----------|-------------|-----------------|
| T-65.1-05-01 | Denial of service | Thread hang on contention | mitigate | Test asserts `t.is_alive() == false` after timeout; wall budget enforced |
| T-65.1-05-02 | Repudiation | Caller cannot distinguish race-loss from real bug | mitigate | Categorisation distinguishes documented races (already_exists, constraint_violation) from unknown_error; full diagnostic text on unknowns |
</threat_model>

<verification>
  <automated>just test-all &gt; /tmp/claude/p05_all.log 2>&amp;1; RC=$?; tail -100 /tmp/claude/p05_all.log; test $RC -eq 0</automated>
</verification>

<success_criteria>
- `test/integration/test_concurrent_writes_per_call_conn.py` exercises N>=4 threads × M>=10 calls
- Categorisation distinguishes known races from unknowns; test fails loud on unknowns
- No hangs; wall budget honored
- Final-state check: no spurious rows in `list_semantic_views()`
- `just test-all` green
</success_criteria>

<output>
Create `.planning/phases/65.1-phase-65-code-review-remediation/65.1-05-SUMMARY.md` when done.
</output>
