fix: backfill session row on out-of-order hooks, add busy_timeout

src/context_engine/memory/db.py

@@ -309,6 +309,11 @@ def connect(db_path: str | Path) -> sqlite3.Connection:
     # WAL gives concurrent readers (the dashboard) decent isolation while the
     # MCP server writes; no impact on single-process use.
     conn.execute("PRAGMA journal_mode = WAL")
+    # Explicitly set the SQLite busy timeout so concurrent writers (hooks +
+    # auto-prune) wait up to 5s for the write lock before raising
+    # "database is locked". This is the SQLite-level PRAGMA, separate from
+    # Python's sqlite3.connect(timeout=...) parameter.
+    conn.execute("PRAGMA busy_timeout = 5000")
     has_vec = _try_load_vec(conn)
     _ensure_schema(conn, has_vec=has_vec)
     return conn

src/context_engine/memory/hooks.py

@@ -47,6 +47,26 @@ def _conn(request: web.Request) -> sqlite3.Connection:
 _RESUME_DECISION_REASON_CHARS = 200
 
 
+def _ensure_session(
+    conn: sqlite3.Connection, session_id: str, project: str = ""
+) -> None:
+    """Ensure a sessions row exists for the given session_id.
+
+    Hooks can arrive out of order (UserPromptSubmit before SessionStart)
+    when cce serve starts mid-session or SessionStart is dropped. Without
+    this, the FOREIGN KEY constraint on prompts/tool_events crashes every
+    subsequent insert for that session. INSERT OR IGNORE is a no-op if the
+    row already exists.
+    """
+    epoch = _now_epoch()
+    conn.execute(
+        "INSERT OR IGNORE INTO sessions "
+        "(id, project, started_at_epoch, started_at, status) "
+        "VALUES (?, ?, ?, ?, 'active')",
+        (session_id, project, epoch, _now_iso(epoch)),
+    )
+
+
 def _build_savings_line(conn: sqlite3.Connection) -> str:
     """One-line savings summary from the savings_log table.
 
@@ -228,6 +248,8 @@ async def handle_user_prompt_submit(request: web.Request) -> web.Response:
 
     conn = _conn(request)
     try:
+        _ensure_session(conn, session_id, request.app.get("project_name", ""))
+
         if prompt_number is None:
             row = conn.execute(
                 "SELECT COALESCE(MAX(prompt_number), 0) + 1 AS next "
@@ -279,6 +301,8 @@ async def handle_post_tool_use(request: web.Request) -> web.Response:
 
     conn = _conn(request)
     try:
+        _ensure_session(conn, session_id, request.app.get("project_name", ""))
+
         if prompt_number is None:
             row = conn.execute(
                 "SELECT COALESCE(MAX(prompt_number), 0) AS cur FROM prompts "
@@ -316,6 +340,8 @@ async def handle_stop(request: web.Request) -> web.Response:
 
     conn = _conn(request)
     try:
+        _ensure_session(conn, session_id, request.app.get("project_name", ""))
+
         if prompt_number is None:
             row = conn.execute(
                 "SELECT COALESCE(MAX(prompt_number), 0) AS cur FROM prompts "
@@ -344,6 +370,8 @@ async def handle_session_end(request: web.Request) -> web.Response:
 
     conn = _conn(request)
     try:
+        _ensure_session(conn, session_id, request.app.get("project_name", ""))
+
         epoch = _now_epoch()
         conn.execute(
             "UPDATE sessions SET status = 'completed', exit_reason = ?, "

tests/memory/test_hooks.py

@@ -418,3 +418,69 @@ async def test_session_start_no_savings_no_line(hook_app, aiohttp_client):
     )
     text = await resp.text()
     assert "CCE saved" not in text
+
+
+# ── Out-of-order hooks (SessionStart missed) ────────────────────────────
+
+
+async def test_prompt_without_session_start_backfills(hook_app, aiohttp_client):
+    """UserPromptSubmit before SessionStart should backfill the sessions row
+    instead of crashing with FOREIGN KEY constraint failed."""
+    app, conn = hook_app
+    client = await aiohttp_client(app)
+    # No SessionStart — go straight to prompt
+    resp = await client.post(
+        "/hooks/UserPromptSubmit",
+        json={"session_id": "orphan", "prompt_text": "hi"},
+    )
+    assert resp.status == 200
+    data = await resp.json()
+    assert data["ok"] is True
+    # Session row was backfilled with project name
+    row = conn.execute("SELECT id, status, project FROM sessions WHERE id = ?", ("orphan",)).fetchone()
+    assert row is not None
+    assert row["status"] == "active"
+    assert row["project"] == "demo"
+
+
+async def test_tool_use_without_session_start_backfills(hook_app, aiohttp_client):
+    """PostToolUse before SessionStart should backfill the sessions row."""
+    app, conn = hook_app
+    client = await aiohttp_client(app)
+    resp = await client.post(
+        "/hooks/PostToolUse",
+        json={
+            "session_id": "orphan2",
+            "tool_name": "Bash",
+            "tool_input": "ls",
+            "tool_output": "file.py",
+        },
+    )
+    assert resp.status == 200
+    data = await resp.json()
+    assert data["ok"] is True
+    row = conn.execute("SELECT id, project FROM sessions WHERE id = ?", ("orphan2",)).fetchone()
+    assert row is not None
+    assert row["project"] == "demo"
+
+
+async def test_stop_without_session_start_does_not_crash(hook_app, aiohttp_client):
+    """Stop before SessionStart should not crash."""
+    app, conn = hook_app
+    client = await aiohttp_client(app)
+    resp = await client.post(
+        "/hooks/Stop",
+        json={"session_id": "orphan3"},
+    )
+    assert resp.status == 200
+
+
+async def test_session_end_without_session_start_does_not_crash(hook_app, aiohttp_client):
+    """SessionEnd before SessionStart should not crash."""
+    app, conn = hook_app
+    client = await aiohttp_client(app)
+    resp = await client.post(
+        "/hooks/SessionEnd",
+        json={"session_id": "orphan4", "exit_reason": "normal"},
+    )
+    assert resp.status == 200