feat(harness): SQLite pending requests for crash-durable approvals

.gitignore

@@ -37,3 +37,7 @@ codex-test-*.png
 # E2E test artifacts
 apps/web/e2e-results/
 test-results/
+
+# Demo scripts (local recording, not shipped)
+apps/web/e2e/demo.spec.ts
+scripts/demo-cli.sh

apps/harness/lib/harness/projector.ex

@@ -109,7 +109,8 @@ defmodule Harness.Projector do
 
   # --- Notification events that resolve pending requests ---
 
-  defp apply_event(snapshot, %Event{kind: :notification, method: "request/resolved"} = event) do
+  defp apply_event(snapshot, %Event{kind: :notification, method: method} = event)
+       when method in ["request/resolved", "user-input/resolved"] do
     request_id = get_in(event.payload, ["requestId"]) || event.event_id
 
     update_session(snapshot, event.thread_id, fn session ->

apps/harness/lib/harness/providers/codex_session.ex

@@ -467,36 +467,42 @@ defmodule Harness.Providers.CodexSession do
   defp assert_codex_version(binary_path, codex_home) do
     env = if codex_home, do: [{"CODEX_HOME", codex_home}], else: []
 
-    try do
-      case System.cmd(binary_path, ["--version"],
-             env: env,
-             stderr_to_stdout: true,
-             timeout: @version_check_timeout_ms
-           ) do
-        {output, 0} ->
-          case parse_codex_version(output) do
-            {:ok, version} ->
-              if version_supported?(version) do
-                :ok
-              else
-                {:error, format_upgrade_message(version)}
-              end
-
-            :error ->
-              {:error,
-               "Could not parse Codex CLI version from output: #{String.slice(output, 0, 200)}"}
-          end
-
-        {output, code} ->
-          {:error,
-           "Codex CLI version check failed (exit #{code}): #{String.slice(output, 0, 200)}"}
-      end
-    rescue
-      e in ErlangError ->
-        case e.original do
-          :enoent -> {:error, "Codex CLI is not installed or not executable at: #{binary_path}"}
-          _ -> {:error, "Codex CLI version check failed: #{inspect(e)}"}
+    task =
+      Task.async(fn ->
+        try do
+          System.cmd(binary_path, ["--version"], env: env, stderr_to_stdout: true)
+        rescue
+          e -> {:rescue, e}
+        end
+      end)
+
+    case Task.yield(task, @version_check_timeout_ms) || Task.shutdown(task) do
+      {:ok, {:rescue, %ErlangError{original: :enoent}}} ->
+        {:error, "Codex CLI is not installed or not executable at: #{binary_path}"}
+
+      {:ok, {:rescue, e}} ->
+        {:error, "Codex CLI version check failed: #{inspect(e)}"}
+
+      {:ok, {output, 0}} ->
+        case parse_codex_version(output) do
+          {:ok, version} ->
+            if version_supported?(version) do
+              :ok
+            else
+              {:error, format_upgrade_message(version)}
+            end
+
+          :error ->
+            {:error,
+             "Could not parse Codex CLI version from output: #{String.slice(output, 0, 200)}"}
         end
+
+      {:ok, {output, code}} ->
+        {:error,
+         "Codex CLI version check failed (exit #{code}): #{String.slice(output, 0, 200)}"}
+
+      nil ->
+        {:error, "Codex CLI version check timed out after #{@version_check_timeout_ms}ms"}
     end
   end
 

apps/harness/lib/harness/snapshot_server.ex

@@ -173,6 +173,12 @@ defmodule Harness.SnapshotServer do
 
     sessions_map = Map.new(sessions, fn session -> {session.thread_id, session} end)
 
+    # Recover pending requests from dedicated table and merge into sessions.
+    # This is the authoritative source — more reliable than the JSON snapshot
+    # in harness_sessions which may be stale if BEAM crashed mid-persist.
+    pending = Harness.Storage.get_pending_requests()
+    sessions_map = merge_pending_into_sessions(sessions_map, pending)
+
     # Set sequence from SQL, NOT from Projector (avoids double-count)
     snapshot = %Snapshot{
       sequence: seq,
@@ -190,6 +196,30 @@ defmodule Harness.SnapshotServer do
     :exit, reason -> {:error, reason}
   end
 
+  defp merge_pending_into_sessions(sessions_map, pending_requests) do
+    # Group pending requests by thread_id, then merge into each session's pending_requests
+    grouped = Enum.group_by(pending_requests, & &1.thread_id)
+
+    Enum.reduce(grouped, sessions_map, fn {thread_id, reqs}, acc ->
+      case Map.get(acc, thread_id) do
+        nil ->
+          acc
+
+        session ->
+          merged_pending =
+            Map.new(reqs, fn req ->
+              {req.request_id, %{
+                method: req.method,
+                payload: req.payload,
+                created_at: req.created_at
+              }}
+            end)
+
+          Map.put(acc, thread_id, %{session | pending_requests: merged_pending})
+      end
+    end)
+  end
+
   @stale_statuses [:running, :connecting]
 
   defp reconcile_stale_sessions(snapshot, seq) do
@@ -300,6 +330,9 @@ defmodule Harness.SnapshotServer do
               })
           end
 
+          # Track pending requests in dedicated table
+          persist_pending_request(event)
+
         {:error, :duplicate_event} ->
           :ok
 
@@ -312,6 +345,29 @@ defmodule Harness.SnapshotServer do
     end
   end
 
+  # INSERT on request/opened, DELETE on request/resolved
+  defp persist_pending_request(%Event{kind: :request} = event) do
+    request_id = get_in(event.payload, ["requestId"]) || event.event_id
+
+    Harness.Storage.insert_pending_request(%{
+      request_id: request_id,
+      thread_id: event.thread_id,
+      provider: event.provider,
+      request_type: get_in(event.payload, ["requestType"]) || "unknown",
+      method: event.method,
+      payload: event.payload,
+      created_at: event.created_at
+    })
+  end
+
+  defp persist_pending_request(%Event{kind: :notification, method: method} = event)
+       when method in ["request/resolved", "user-input/resolved"] do
+    request_id = get_in(event.payload, ["requestId"]) || event.event_id
+    Harness.Storage.resolve_pending_request(request_id)
+  end
+
+  defp persist_pending_request(_event), do: :ok
+
   # --- SQL replay fallback ---
 
   defp replay_from_sql_or_empty(after_seq, seq, state) do

apps/harness/lib/harness/storage.ex

@@ -49,6 +49,21 @@ defmodule Harness.Storage do
     GenServer.call(__MODULE__, :get_event_count)
   end
 
+  @doc "Insert a pending request (approval or elicitation). Idempotent on request_id."
+  def insert_pending_request(request_map) do
+    GenServer.call(__MODULE__, {:insert_pending_request, request_map})
+  end
+
+  @doc "Mark a pending request as resolved. Deletes the row."
+  def resolve_pending_request(request_id) do
+    GenServer.call(__MODULE__, {:resolve_pending_request, request_id})
+  end
+
+  @doc "Get all unresolved pending requests, optionally filtered by thread_id."
+  def get_pending_requests(thread_id \\ nil) do
+    GenServer.call(__MODULE__, {:get_pending_requests, thread_id})
+  end
+
   @doc "Truncate all tables. Test-only."
   def reset! do
     GenServer.call(__MODULE__, :reset)
@@ -101,7 +116,23 @@ defmodule Harness.Storage do
     {:reply, count, state}
   end
 
+  def handle_call({:insert_pending_request, request_map}, _from, %{conn: conn} = state) do
+    result = do_insert_pending_request(conn, request_map)
+    {:reply, result, state}
+  end
+
+  def handle_call({:resolve_pending_request, request_id}, _from, %{conn: conn} = state) do
+    result = do_resolve_pending_request(conn, request_id)
+    {:reply, result, state}
+  end
+
+  def handle_call({:get_pending_requests, thread_id}, _from, %{conn: conn} = state) do
+    result = do_get_pending_requests(conn, thread_id)
+    {:reply, result, state}
+  end
+
   def handle_call(:reset, _from, %{conn: conn} = state) do
+    Sqlite3.execute(conn, "DELETE FROM harness_pending_requests")
     Sqlite3.execute(conn, "DELETE FROM harness_events")
     Sqlite3.execute(conn, "DELETE FROM harness_sessions")
     {:reply, :ok, state}
@@ -180,6 +211,25 @@ defmodule Harness.Storage do
       last_sequence INTEGER NOT NULL DEFAULT 0
     )
     """)
+
+    # Pending requests: approvals + elicitations that survive BEAM restarts.
+    # Rows are INSERTed on request/opened and DELETEd on request/resolved.
+    Sqlite3.execute(conn, """
+    CREATE TABLE IF NOT EXISTS harness_pending_requests (
+      request_id TEXT PRIMARY KEY,
+      thread_id TEXT NOT NULL,
+      provider TEXT NOT NULL,
+      request_type TEXT NOT NULL,
+      method TEXT NOT NULL,
+      payload_json TEXT NOT NULL,
+      created_at TEXT NOT NULL
+    )
+    """)
+
+    Sqlite3.execute(conn, """
+    CREATE INDEX IF NOT EXISTS idx_pending_reqs_thread
+      ON harness_pending_requests(thread_id)
+    """)
   end
 
   # --- Event operations ---
@@ -282,6 +332,74 @@ defmodule Harness.Storage do
     end
   end
 
+  # --- Pending request operations ---
+
+  defp do_insert_pending_request(conn, req) do
+    sql = """
+    INSERT OR IGNORE INTO harness_pending_requests
+      (request_id, thread_id, provider, request_type, method, payload_json, created_at)
+    VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
+    """
+
+    payload_json = if req[:payload], do: Jason.encode!(req.payload), else: "{}"
+
+    params = [
+      req.request_id,
+      req.thread_id,
+      req.provider,
+      req[:request_type] || "unknown",
+      req[:method] || "request/opened",
+      payload_json,
+      req[:created_at] || DateTime.utc_now() |> DateTime.to_iso8601()
+    ]
+
+    {:ok, stmt} = Sqlite3.prepare(conn, sql)
+
+    try do
+      :ok = Sqlite3.bind(stmt, params)
+      :done = Sqlite3.step(conn, stmt)
+      :ok
+    after
+      Sqlite3.release(conn, stmt)
+    end
+  end
+
+  defp do_resolve_pending_request(conn, request_id) do
+    sql = "DELETE FROM harness_pending_requests WHERE request_id = ?1"
+    {:ok, stmt} = Sqlite3.prepare(conn, sql)
+
+    try do
+      :ok = Sqlite3.bind(stmt, [request_id])
+      :done = Sqlite3.step(conn, stmt)
+      :ok
+    after
+      Sqlite3.release(conn, stmt)
+    end
+  end
+
+  defp do_get_pending_requests(conn, nil) do
+    sql = """
+    SELECT request_id, thread_id, provider, request_type, method, payload_json, created_at
+    FROM harness_pending_requests
+    ORDER BY created_at ASC
+    """
+
+    query_all(conn, sql, [])
+    |> Enum.map(&row_to_pending_request/1)
+  end
+
+  defp do_get_pending_requests(conn, thread_id) do
+    sql = """
+    SELECT request_id, thread_id, provider, request_type, method, payload_json, created_at
+    FROM harness_pending_requests
+    WHERE thread_id = ?1
+    ORDER BY created_at ASC
+    """
+
+    query_all(conn, sql, [thread_id])
+    |> Enum.map(&row_to_pending_request/1)
+  end
+
   # --- Session queries ---
 
   defp do_get_all_sessions(conn) do
@@ -414,6 +532,18 @@ defmodule Harness.Storage do
     }
   end
 
+  defp row_to_pending_request([request_id, thread_id, provider, request_type, method, payload_json, created_at]) do
+    %{
+      request_id: request_id,
+      thread_id: thread_id,
+      provider: provider,
+      request_type: request_type,
+      method: method,
+      payload: decode_json(payload_json) || %{},
+      created_at: created_at
+    }
+  end
+
   # --- Serialization helpers ---
 
   defp serialize_atom(nil), do: nil