[P6.6.1] Add busy_timeout + WAL to core workspace SQLite connections

[frankbria]

Problem

The core workspace SQLite connections set no busy_timeout and no WAL journaling. Under parallel batch execution, multiple subprocesses write to the same workspace DB concurrently, and server background agent threads add more concurrent writers. With no busy_timeout, a concurrent writer gets an immediate database is locked OperationalError instead of waiting.

The platform_store Database already sets WAL + busy_timeout=5000, so core is inconsistent with the rest of the codebase.

Evidence

• codeframe/core/workspace.py:839 (get_db_connection) and the other direct sqlite3.connect() calls use bare defaults
• codeframe/core/conductor.py:2087+ (_execute_task_subprocess) — parallel writers
• codeframe/ui/routers/tasks_v2.py:716 — server-triggered concurrent agent threads
• codeframe/platform_store/database.py:78-84 — sets WAL + busy_timeout=5000 (the pattern core should match)

Fix

• In get_db_connection (and the other direct connects), set PRAGMA busy_timeout and PRAGMA journal_mode=WAL, matching platform_store/database.py.

Acceptance criteria

• Core workspace connections set busy_timeout and WAL.
• A concurrency test (parallel writers) no longer raises immediate database is locked.

Source: release-readiness audit 2026-06-13 (backend agent). Top backend fix before parallel batch execution is relied upon in production.

[coderabbitai]

Coding Plan

Summary

• Introduce a single private helper _open_db() in workspace.py that applies WAL + busy_timeout, avoiding duplication across 10 call sites
• Refactor workspace.py internals and external callers (costs_v2.py, budget.py) to route through this centralized connection setup
• Add a threading-based concurrency test to verify the fix prevents immediate lock errors under parallel write load

Implementation Steps

Phase 1: Add Connection Helper and Update Workspace Module

Introduce a private helper function in workspace.py that applies WAL mode and busy_timeout to all connections, then refactor all six direct sqlite3.connect() calls in the module to use this helper.

Task 1: Create Private Connection Helper

Add a private function that encapsulates the connection setup pattern from platform_store/database.py.

• Add _open_db(db_path: str | Path) -> sqlite3.Connection function near the top of codeframe/core/workspace.py
• Function should call sqlite3.connect(str(db_path)), then execute PRAGMA journal_mode = WAL and PRAGMA busy_timeout = 5000 before returning the connection
• Keep it simple—no row_factory or foreign_keys (matching existing workspace patterns)

Task 2: Update Internal Workspace Functions

Refactor the five internal functions that manage their own connection lifecycle to use the new helper.

• Update _init_database (line 73) to use _open_db(db_path) instead of sqlite3.connect(db_path)
• Update _ensure_schema_upgrades (line 411) to use _open_db(db_path)
• Update create_or_load_workspace (line 770) to use _open_db(db_path)
• Update get_workspace (line 810) to use _open_db(db_path)
• Update update_workspace_tech_stack (line 878) to use _open_db(workspace.db_path)

Task 3: Update Public get_db_connection Function

Update the public-facing connection factory to use the helper, automatically fixing all external callers that use it.

• Update get_db_connection (line 839) to use _open_db(workspace.db_path) instead of direct sqlite3.connect
• Docstring remains unchanged—caller is still responsible for closing

🤖 Prompt for AI agents

Implement a private connection helper in `codeframe/core/workspace.py` and
refactor all six direct `sqlite3.connect()` calls in the module to use it.

- Near the top of `codeframe/core/workspace.py`, add a private function
`_open_db(db_path: str | Path) -> sqlite3.Connection`:
  - Call `sqlite3.connect(str(db_path))`
  - Execute `PRAGMA journal_mode = WAL` on the connection
  - Execute `PRAGMA busy_timeout = 5000` on the connection
  - Return the connection
  - Do NOT add `row_factory` or `foreign_keys` settings
- Replace `sqlite3.connect(db_path)` in `_init_database` (line 73) with
`_open_db(db_path)`
- Replace `sqlite3.connect(db_path)` in `_ensure_schema_upgrades` (line 411)
with `_open_db(db_path)`
- Replace `sqlite3.connect(db_path)` in `create_or_load_workspace` (line 770)
with `_open_db(db_path)`
- Replace `sqlite3.connect(db_path)` in `get_workspace` (line 810) with
`_open_db(db_path)`
- Replace `sqlite3.connect(workspace.db_path)` in `update_workspace_tech_stack`
(line 878) with `_open_db(workspace.db_path)`
- Replace `sqlite3.connect(...)` in `get_db_connection` (line 839) with
`_open_db(workspace.db_path)`; leave the docstring unchanged

Phase 2: Update External Callers to Use Centralized Helper

Migrate the remaining direct sqlite3.connect() calls in costs_v2.py and budget.py to use the workspace module's get_db_connection function, ensuring all workspace database access goes through the properly-configured path.

Task 1: Update costs_v2.py Connection Handling

Replace direct sqlite3.connect() calls with get_db_connection from the workspace module.

• Add import for get_db_connection from codeframe.core.workspace in codeframe/ui/routers/costs_v2.py
• Update _query_costs (line 82) to use get_db_connection(workspace) instead of sqlite3.connect(db_path)
• Update _open_workspace_conn helper (line 187) to use get_db_connection(workspace) instead of sqlite3.connect(db_path)
• Retain existing conn.row_factory = sqlite3.Row assignments after obtaining the connection

Task 2: Update budget.py Connection Handling

Replace direct sqlite3.connect() calls with get_db_connection from the workspace module.

• Add import for get_db_connection from codeframe.core.workspace in codeframe/adapters/e2b/budget.py
• Update record_cloud_run (line 35) to use get_db_connection(workspace) instead of sqlite3.connect(workspace.db_path)
• Update get_cloud_run (line 63) to use get_db_connection(workspace) and add conn.row_factory = sqlite3.Row after obtaining connection

🤖 Prompt for AI agents

Migrate all remaining direct `sqlite3.connect()` calls in `costs_v2.py` and
`budget.py` to use the `get_db_connection` function from the workspace module.

**`codeframe/ui/routers/costs_v2.py`:**
- Add `from codeframe.core.workspace import get_db_connection` to the imports
- In `_query_costs` (line 82), replace `sqlite3.connect(db_path)` with
`get_db_connection(workspace)`
- In `_open_workspace_conn` (line 187), replace `sqlite3.connect(db_path)` with
`get_db_connection(workspace)`
- Keep any existing `conn.row_factory = sqlite3.Row` assignments in place after
obtaining the connection

**`codeframe/adapters/e2b/budget.py`:**
- Add `from codeframe.core.workspace import get_db_connection` to the imports
- In `record_cloud_run` (line 35), replace `sqlite3.connect(workspace.db_path)`
with `get_db_connection(workspace)`
- In `get_cloud_run` (line 63), replace `sqlite3.connect(workspace.db_path)`
with `get_db_connection(workspace)` and add `conn.row_factory = sqlite3.Row`
immediately after obtaining the connection

Phase 3: Add Concurrency Test

Create a test that verifies parallel writers to the workspace database no longer receive immediate "database is locked" errors.

Create Parallel Writer Test

Add a test that spawns concurrent database writers and confirms the busy_timeout allows them to complete without OperationalError.

• Create test in tests/core/test_workspace.py following existing patterns with tmp_path fixture and create_or_load_workspace()
• Use concurrent.futures.ThreadPoolExecutor to spawn 5–10 parallel threads, each performing INSERT operations via get_db_connection
• Each thread should perform multiple writes with small delays to create contention
• Assert that all threads complete successfully without raising sqlite3.OperationalError
• Add pytestmark = pytest.mark.v2 if following costs_v2.py test conventions

🤖 Prompt for AI agents

Add a concurrency test to `tests/core/test_workspace.py` that verifies parallel
database writers do not encounter immediate "database is locked" errors.

- Follow existing test patterns in the file using the `tmp_path` pytest fixture
and `create_or_load_workspace()`
- Use `concurrent.futures.ThreadPoolExecutor` to spawn 5–10 parallel threads
- Each thread should obtain a connection via `get_db_connection`, perform INSERT
operations, and include small delays to create write contention between threads
- Assert that all threads complete successfully and that no
`sqlite3.OperationalError` is raised
- Add `pytestmark = pytest.mark.v2` at the top of the test if that convention is
followed in `costs_v2.py` tests

Research

The codebase has two distinct SQLite usage patterns: platform_store/database.py uses a centralized connection manager with WAL mode and a 5-second busy timeout for concurrent access protection, while codeframe/core/workspace.py and its consumers use bare sqlite3.connect() calls without any concurrency safeguards. There are 10 total direct connection calls needing updates: 6 in workspace.py, 2 in costs_v2.py, and 2 in budget.py. The conductor.py and tasks_v2.py files mentioned in the ticket do not directly open SQLite connections—they trigger concurrent execution but the actual DB access happens through other paths.

---

🚀 Next Steps

🤖 All AI agent prompts combined

Task: 1

Implement a private connection helper in `codeframe/core/workspace.py` and
refactor all six direct `sqlite3.connect()` calls in the module to use it.

- Near the top of `codeframe/core/workspace.py`, add a private function
`_open_db(db_path: str | Path) -> sqlite3.Connection`:
  - Call `sqlite3.connect(str(db_path))`
  - Execute `PRAGMA journal_mode = WAL` on the connection
  - Execute `PRAGMA busy_timeout = 5000` on the connection
  - Return the connection
  - Do NOT add `row_factory` or `foreign_keys` settings
- Replace `sqlite3.connect(db_path)` in `_init_database` (line 73) with
`_open_db(db_path)`
- Replace `sqlite3.connect(db_path)` in `_ensure_schema_upgrades` (line 411)
with `_open_db(db_path)`
- Replace `sqlite3.connect(db_path)` in `create_or_load_workspace` (line 770)
with `_open_db(db_path)`
- Replace `sqlite3.connect(db_path)` in `get_workspace` (line 810) with
`_open_db(db_path)`
- Replace `sqlite3.connect(workspace.db_path)` in `update_workspace_tech_stack`
(line 878) with `_open_db(workspace.db_path)`
- Replace `sqlite3.connect(...)` in `get_db_connection` (line 839) with
`_open_db(workspace.db_path)`; leave the docstring unchanged
===============================================================================

Task: 2

Migrate all remaining direct `sqlite3.connect()` calls in `costs_v2.py` and
`budget.py` to use the `get_db_connection` function from the workspace module.

**`codeframe/ui/routers/costs_v2.py`:**
- Add `from codeframe.core.workspace import get_db_connection` to the imports
- In `_query_costs` (line 82), replace `sqlite3.connect(db_path)` with
`get_db_connection(workspace)`
- In `_open_workspace_conn` (line 187), replace `sqlite3.connect(db_path)` with
`get_db_connection(workspace)`
- Keep any existing `conn.row_factory = sqlite3.Row` assignments in place after
obtaining the connection

**`codeframe/adapters/e2b/budget.py`:**
- Add `from codeframe.core.workspace import get_db_connection` to the imports
- In `record_cloud_run` (line 35), replace `sqlite3.connect(workspace.db_path)`
with `get_db_connection(workspace)`
- In `get_cloud_run` (line 63), replace `sqlite3.connect(workspace.db_path)`
with `get_db_connection(workspace)` and add `conn.row_factory = sqlite3.Row`
immediately after obtaining the connection
===============================================================================

Task: 3

Add a concurrency test to `tests/core/test_workspace.py` that verifies parallel
database writers do not encounter immediate "database is locked" errors.

- Follow existing test patterns in the file using the `tmp_path` pytest fixture
and `create_or_load_workspace()`
- Use `concurrent.futures.ThreadPoolExecutor` to spawn 5–10 parallel threads
- Each thread should obtain a connection via `get_db_connection`, perform INSERT
operations, and include small delays to create write contention between threads
- Assert that all threads complete successfully and that no
`sqlite3.OperationalError` is raised
- Add `pytestmark = pytest.mark.v2` at the top of the test if that convention is
followed in `costs_v2.py` tests

💡 Iterate on the plan with: `@coderabbitai` <feedback>

Example Feedback
- `@coderabbitai` You can skip phase 3. Add a simple unit test case for phase 2.
- `@coderabbitai` For design choice 1 go ahead with option 3 and replan.

---

💬 Have feedback or questions? Drop into our discord!

[coderabbitai]

🔗 Related PRs

#229 - feat(e2e): add state reconciliation tests for late-joining users [merged]
#604 - refactor: rename persistence -> platform_store + v1 cleanup (#599 PR3/3) [merged]

---

📝 Issue Planner

_{Check the box below or use the @coderabbitai plan command to generate an implementation plan and prompts that you can use with your favorite coding assistant.}

• Create Plan

---

🧪 Issue enrichment is currently in open beta.

To disable automatic issue enrichment, add the following to your .coderabbit.yaml:

issue_enrichment:
  auto_enrich:
    enabled: false

💬 Have feedback or questions? Drop into our discord!