Close TD-17 language detection coverage

TECHDEBTS.md

@@ -518,7 +518,17 @@ debounced save. Queries during a rebuild keep hitting the old index.
 **Acceptance.** Neighbor queries return while a duplicate scan is running
 (test with a large synthetic index); debounce/flush semantics unchanged.
 
-### TD-13: Slim down the `Publishing` GenServer call surface
+### TD-13: Slim down the `Publishing` GenServer call surface ✅ DONE (2026-06-12)
+
+**Status: implemented.** `BDS.Publishing` now keeps only SCP mtime state in its
+GenServer. Publish-job creation, lookup, and status updates run directly through
+`Repo`, while background-task job updates use a stable Repo caller so sandboxed
+tests still exercise the real async path. SCP uploads no longer round-trip
+through `should_upload_scp_file` / `mark_uploaded_scp_file` per file; each
+target now batches one filter call for changed files and one bulk record call
+for successfully uploaded mtimes. Coverage includes a focused batching test that
+proves a multi-file SCP publish keeps bookkeeping traffic bounded instead of
+scaling linearly with file count.
 
 **Context.** `Publishing` does Repo writes inside `handle_call`
 (`:upload_site`, `:update_job`) and the uploader makes per-file synchronous
@@ -537,7 +547,20 @@ file. Consider whether `scp_uploads` state should be ETS
 **Acceptance.** Upload of N files makes O(1) GenServer calls for mtime
 bookkeeping, not O(N)·2; behavior identical for incremental uploads.
 
-### TD-14: Replace polling with messaging (CliSync watcher + rebuild sequencing)
+### TD-14: Replace polling with messaging (CliSync watcher + rebuild sequencing) ✅ DONE (2026-06-12)
+
+**Status: implemented.** `BDS.CliSync.Watcher` now gates notification-table
+work behind SQLite `PRAGMA data_version`, so unchanged databases no longer force
+repeated `db_notifications` queries at the 100 ms watch cadence; the watcher
+still performs a real notification fetch/prune pass on the first poll and on
+every external commit boundary. Rebuild sequencing in
+`BDS.Desktop.ShellCommands` no longer sleep-polls `Tasks.list_tasks/0`.
+`BDS.Tasks` now broadcasts terminal task states on `BDS.PubSub`, and
+`wait_for_group_phase/3` subscribes and waits on those messages with the same
+deadline semantics. Coverage now includes a watcher test that proves unchanged
+`data_version` skips notification queries, a tasks test that proves terminal
+state broadcasts, and a shell-command guard test that forbids `Process.sleep`
+polling in the rebuild wait path.
 
 **Context.** Two polling loops:
 1. `CliSync.Watcher` polls the SQLite notifications table every **100 ms
@@ -567,7 +590,16 @@ bookkeeping, not O(N)·2; behavior identical for incremental uploads.
 unchanged (or interval ≥ 1s with backoff); rebuild sequencing has no
 `Process.sleep`; CLI-sync round-trip latency stays ≤ current behavior.
 
-### TD-15: `BDS.Tasks` housekeeping (queue type, eviction timers)
+### TD-15: `BDS.Tasks` housekeeping (queue type, eviction timers) ✅ DONE (2026-06-12)
+
+**Status: implemented.** `BDS.Tasks` now uses `:queue` for its pending work
+queue, so enqueue/dequeue on the hot path are O(1) and the FIFO behavior is
+unchanged. Finished-task cleanup now tracks a single live eviction timer ref
+instead of scheduling a fresh `send_after/3` on every terminal task; the timer
+fires, prunes expired finished tasks, and only reschedules itself if finished
+tasks still remain. Coverage now includes a focused task-state test proving
+multiple finished tasks share the same live eviction timer and a source guard
+that forbids `queue ++` churn.
 
 **Context.** Minor inefficiencies in `tasks.ex`: the pending queue is a list
 appended with `++` (O(n) per submit), and **every** finishing task schedules
@@ -590,7 +622,18 @@ existing task lifecycle tests green.
 
 ## Phase 4 — Build-vs-buy replacements
 
-### TD-16: Frontmatter robustness — yaml_elixir/ymlr or harden the hand-rolled parser
+### TD-16: Frontmatter robustness — yaml_elixir/ymlr or harden the hand-rolled parser ✅ DONE (2026-06-12)
+
+**Status: implemented.** The project keeps the hand-rolled serializer/parser for
+byte-stable frontmatter output, but `BDS.Frontmatter` is now hardened for the
+known user-edited-file cases: `parse_document/1` normalizes CRLF and lone `\r`
+line endings before splitting the gray-matter envelope, and quoted scalar
+parsing now removes exactly one closing quote and unescapes content explicitly
+instead of `trim_trailing/2`, which previously corrupted strings whose content
+ended in a quote character. Coverage now includes focused parser tests for CRLF
+documents and quoted-string roundtrips with embedded quotes, backslashes, and
+trailing-quote content, plus adjacent post/template/maintenance frontmatter and
+serializer parity suites.
 
 **Context.** `BDS.Frontmatter` is a hand-rolled YAML subset with concrete
 bugs for user-edited files:
@@ -621,7 +664,16 @@ user projects.
 **Acceptance.** CRLF fixture parses; round-trip property tests pass; golden
 serialization fixtures unchanged (if keeping custom serializer).
 
-### TD-17: Language detection via `paasaa` (optional, low priority)
+### TD-17: Language detection via `paasaa` (optional, low priority) ✅ DONE (2026-06-12)
+
+**Status: implemented without adding `paasaa`.** The originally reported
+misclassifications are not reproducible on the current code: the existing
+detector already classifies the relevant umlaut-free German and accent-free
+French fixtures correctly through its language-hint fallback, and new focused
+tests now lock that behavior down directly. Because the acceptance cases are now
+satisfied and the current implementation keeps dependency weight lower, the
+project does not add `paasaa` at this time. Revisit only if broader real-world
+fixtures start failing.
 
 **Context.** `Search.detect_language/1` uses diacritic regexes + tiny word
 lists; German text without umlauts (common in short posts) falls through to

lib/bds/cli_sync.ex

@@ -54,6 +54,11 @@ defmodule BDS.CliSync do
      end)}
   end
 
+  def data_version do
+    %{rows: [[version]]} = Repo.query!("PRAGMA data_version", [])
+    version
+  end
+
   def prune_notifications(now \\ Persistence.now_ms()) when is_integer(now) do
     {processed_count, _} =
       Repo.delete_all(

lib/bds/cli_sync/watcher.ex

@@ -29,7 +29,12 @@ defmodule BDS.CliSync.Watcher do
           Keyword.get(opts, :poll_interval_ms),
           @default_poll_interval_ms
         ),
-      pubsub: Keyword.get(opts, :pubsub, BDS.PubSub)
+      pubsub: Keyword.get(opts, :pubsub, BDS.PubSub),
+      data_version_reader: Keyword.get(opts, :data_version_reader, &CliSync.data_version/0),
+      notification_fetcher:
+        Keyword.get(opts, :notification_fetcher, &CliSync.db_file_change_detected/0),
+      pruner: Keyword.get(opts, :pruner, &CliSync.prune_notifications/0),
+      last_data_version: nil
     }
 
     {:ok, schedule_poll(state)}
@@ -49,18 +54,24 @@ defmodule BDS.CliSync.Watcher do
   end
 
   defp process_notifications(state) do
-    {:ok, notifications} = CliSync.db_file_change_detected()
-    {:ok, _pruned} = CliSync.prune_notifications()
+    current_data_version = state.data_version_reader.()
 
-    Enum.each(notifications, fn notification ->
-      Phoenix.PubSub.broadcast(
-        state.pubsub,
-        topic(),
-        {:entity_changed, notification_payload(notification)}
-      )
-    end)
+    if state.last_data_version == current_data_version do
+      %{state | last_data_version: current_data_version}
+    else
+      {:ok, notifications} = state.notification_fetcher.()
+      {:ok, _pruned} = state.pruner.()
 
-    state
+      Enum.each(notifications, fn notification ->
+        Phoenix.PubSub.broadcast(
+          state.pubsub,
+          topic(),
+          {:entity_changed, notification_payload(notification)}
+        )
+      end)
+
+      %{state | last_data_version: current_data_version}
+    end
   end
 
   defp notification_payload(notification) do

lib/bds/desktop/shell_commands.ex

@@ -559,27 +559,62 @@ defmodule BDS.Desktop.ShellCommands do
     end
   end
 
-  defp wait_for_group_phase(_group_id, _names, timeout) when timeout <= 0, do: :timeout
-
   defp wait_for_group_phase(group_id, names, timeout) do
+    if timeout <= 0 do
+      :timeout
+    else
+      Phoenix.PubSub.subscribe(BDS.PubSub, Tasks.topic())
+
+      try do
+        case group_phase_status(group_id, names) do
+          :waiting -> wait_for_group_phase_message(group_id, names, timeout)
+          status -> status
+        end
+      after
+        Phoenix.PubSub.unsubscribe(BDS.PubSub, Tasks.topic())
+      end
+    end
+  end
+
+  defp wait_for_group_phase_message(group_id, names, timeout) do
+    started_at = System.monotonic_time(:millisecond)
+
+    receive do
+      {:task_terminal, task} ->
+        elapsed = System.monotonic_time(:millisecond) - started_at
+
+        cond do
+          task.group_id == group_id and task.name in names and task.status == :failed ->
+            :failed
+
+          task.group_id == group_id and task.name in names ->
+            case group_phase_status(group_id, names) do
+              :waiting ->
+                wait_for_group_phase_message(group_id, names, timeout - elapsed)
+
+              status ->
+                status
+            end
+
+          true ->
+            wait_for_group_phase_message(group_id, names, timeout - elapsed)
+        end
+    after
+      timeout ->
+        :timeout
+    end
+  end
+
+  defp group_phase_status(group_id, names) do
     tasks =
       BDS.Tasks.list_tasks()
       |> Enum.filter(&(&1.group_id == group_id and &1.name in names))
 
     cond do
-      length(tasks) < length(names) ->
-        Process.sleep(50)
-        wait_for_group_phase(group_id, names, timeout - 50)
-
-      Enum.any?(tasks, &(&1.status == :failed)) ->
-        :failed
-
-      Enum.all?(tasks, &(&1.status == :completed)) ->
-        :ok
-
-      true ->
-        Process.sleep(50)
-        wait_for_group_phase(group_id, names, timeout - 50)
+      length(tasks) < length(names) -> :waiting
+      Enum.any?(tasks, &(&1.status == :failed)) -> :failed
+      Enum.all?(tasks, &(&1.status == :completed)) -> :ok
+      true -> :waiting
     end
   end
 

lib/bds/frontmatter.ex

@@ -17,7 +17,9 @@ defmodule BDS.Frontmatter do
   end
 
   def parse_document(contents) when is_binary(contents) do
-    case String.split(contents, "\n---\n", parts: 2) do
+    normalized_contents = normalize_newlines(contents)
+
+    case String.split(normalized_contents, "\n---\n", parts: 2) do
       [frontmatter_with_marker, body] ->
         frontmatter = String.replace_prefix(frontmatter_with_marker, "---\n", "")
 
@@ -163,19 +165,11 @@ defmodule BDS.Frontmatter do
   end
 
   defp parse_string("\"" <> rest) do
-    rest
-    |> String.trim_trailing("\"")
-    |> String.replace("\\n", "\n")
-    |> String.replace("\\\"", "\"")
-    |> String.replace("\\\\", "\\")
+    parse_quoted_string(rest, ?")
   end
 
   defp parse_string("'" <> rest) do
-    rest
-    |> String.trim_trailing("'")
-    |> String.replace("\\n", "\n")
-    |> String.replace("\\'", "'")
-    |> String.replace("\\\\", "\\")
+    parse_quoted_string(rest, ?')
   end
 
   defp parse_string(value), do: value
@@ -235,4 +229,46 @@ defmodule BDS.Frontmatter do
     rendered = to_string(key)
     String.ends_with?(rendered, "_at") or String.ends_with?(rendered, "At")
   end
+
+  defp normalize_newlines(contents) do
+    contents
+    |> String.replace("\r\n", "\n")
+    |> String.replace("\r", "\n")
+  end
+
+  defp parse_quoted_string(rest, quote) do
+    quote_binary = <<quote::utf8>>
+
+    if String.ends_with?(rest, quote_binary) do
+      inner = binary_part(rest, 0, byte_size(rest) - byte_size(quote_binary))
+      unescape_quoted_string(inner, quote, "")
+    else
+      quote_binary <> rest
+    end
+  end
+
+  defp unescape_quoted_string(<<>>, _quote, acc), do: acc
+
+  defp unescape_quoted_string("\\" <> rest, quote, acc) do
+    case rest do
+      <<"n", tail::binary>> ->
+        unescape_quoted_string(tail, quote, acc <> "\n")
+
+      <<"\\", tail::binary>> ->
+        unescape_quoted_string(tail, quote, acc <> "\\")
+
+      <<escaped, tail::binary>> when escaped == quote ->
+        unescape_quoted_string(tail, quote, acc <> <<quote::utf8>>)
+
+      <<char::utf8, tail::binary>> ->
+        unescape_quoted_string(tail, quote, acc <> "\\" <> <<char::utf8>>)
+
+      <<>> ->
+        acc <> "\\"
+    end
+  end
+
+  defp unescape_quoted_string(<<char::utf8, tail::binary>>, quote, acc) do
+    unescape_quoted_string(tail, quote, acc <> <<char::utf8>>)
+  end
 end

lib/bds/publishing.ex

@@ -28,52 +28,7 @@ defmodule BDS.Publishing do
     project = Projects.get_project!(project_id)
     normalized_credentials = normalize_credentials(credentials)
     targets = build_upload_targets(Projects.project_data_dir(project), normalized_credentials)
-    GenServer.call(__MODULE__, {:upload_site, project_id, normalized_credentials, targets, opts})
-  end
 
-  @spec get_job(String.t()) :: PublishJob.t() | nil
-  def get_job(job_id) when is_binary(job_id) do
-    GenServer.call(__MODULE__, {:get_job, job_id})
-  end
-
-  @impl true
-  def init(_state) do
-    {:ok, %{scp_uploads: %{}}}
-  end
-
-  @impl true
-  def handle_call({:get_job, job_id}, _from, state) do
-    {:reply, Repo.get(PublishJob, job_id), state}
-  end
-
-  @impl true
-  def handle_call({:update_job, job_id, attrs}, _from, state) do
-    with %PublishJob{} = job <- Repo.get(PublishJob, job_id) do
-      attrs = Map.put(attrs, :updated_at, Persistence.now_ms())
-      job |> PublishJob.changeset(attrs) |> Repo.update()
-    end
-
-    {:reply, :ok, state}
-  end
-
-  @impl true
-  def handle_call({:should_upload_scp_file, upload_key, local_mtime}, _from, state) do
-    should_upload? =
-      case state.scp_uploads[upload_key] do
-        nil -> true
-        recorded_mtime -> local_mtime > recorded_mtime
-      end
-
-    {:reply, should_upload?, state}
-  end
-
-  @impl true
-  def handle_call({:mark_uploaded_scp_file, upload_key, local_mtime}, _from, state) do
-    {:reply, :ok, put_in(state, [:scp_uploads, upload_key], local_mtime)}
-  end
-
-  @impl true
-  def handle_call({:upload_site, project_id, credentials, targets, opts}, _from, state) do
     job_id = "publish-" <> Integer.to_string(System.unique_integer([:positive, :monotonic]))
     uploader = build_uploader(Keyword.put_new(opts, :project_id, project_id))
     now = Persistence.now_ms()
@@ -83,10 +38,10 @@ defmodule BDS.Publishing do
       project_id: project_id,
       status: :pending,
       task_id: nil,
-      ssh_host: credentials.ssh_host,
-      ssh_user: credentials.ssh_user,
-      ssh_remote_path: credentials.ssh_remote_path,
-      ssh_mode: credentials.ssh_mode,
+      ssh_host: normalized_credentials.ssh_host,
+      ssh_user: normalized_credentials.ssh_user,
+      ssh_remote_path: normalized_credentials.ssh_remote_path,
+      ssh_mode: normalized_credentials.ssh_mode,
       targets: Enum.map(targets, &to_string(&1.kind)),
       error: nil,
       inserted_at: now,
@@ -102,7 +57,7 @@ defmodule BDS.Publishing do
       Tasks.submit_task(
         "publish #{project_id}",
         fn report ->
-          run_upload(job_id, credentials, targets, uploader, report)
+          run_upload(job_id, normalized_credentials, targets, uploader, report)
         end,
         %{
           group_id: project_id,
@@ -115,7 +70,51 @@ defmodule BDS.Publishing do
       |> PublishJob.changeset(%{task_id: task.id, updated_at: Persistence.now_ms()})
       |> Repo.update!()
 
-    {:reply, {:ok, next_job}, state}
+    {:ok, next_job}
+  end
+
+  @spec get_job(String.t()) :: PublishJob.t() | nil
+  def get_job(job_id) when is_binary(job_id) do
+    Repo.get(PublishJob, job_id)
+  end
+
+  @impl true
+  def init(_state) do
+    {:ok, %{scp_uploads: %{}}}
+  end
+
+  @impl true
+  def handle_call({:filter_scp_uploads, project_id, credentials, target_kind, files}, _from, state) do
+    {files_to_upload, next_uploads} =
+      Enum.reduce(files, {[], state.scp_uploads}, fn {relative_path, local_mtime}, {acc, uploads} ->
+        upload_key = scp_upload_key(project_id, credentials, target_kind, relative_path)
+
+        if should_upload_mtime?(uploads[upload_key], local_mtime) do
+          {[{relative_path, local_mtime} | acc], uploads}
+        else
+          {acc, uploads}
+        end
+      end)
+
+    {:reply, Enum.reverse(files_to_upload), %{state | scp_uploads: next_uploads}}
+  end
+
+  @impl true
+  def handle_call(
+        {:record_uploaded_scp_files, project_id, credentials, target_kind, uploaded_files},
+        _from,
+        state
+      ) do
+    next_uploads =
+      Enum.reduce(uploaded_files, state.scp_uploads, fn {relative_path, local_mtime}, uploads ->
+        Map.put(
+          uploads,
+          scp_upload_key(project_id, credentials, target_kind, relative_path),
+          local_mtime
+        )
+      end)
+
+    {:reply, :ok, %{state | scp_uploads: next_uploads}}
   end
 
   defp run_upload(job_id, credentials, targets, uploader, report) do
@@ -147,7 +146,14 @@ defmodule BDS.Publishing do
   end
 
   defp update_job(job_id, attrs) do
-    GenServer.call(__MODULE__, {:update_job, job_id, attrs})
+    repo_opts = repo_call_opts()
+
+    with %PublishJob{} = job <- Repo.get(PublishJob, job_id, repo_opts) do
+      attrs = Map.put(attrs, :updated_at, Persistence.now_ms())
+      _ = job |> PublishJob.changeset(attrs) |> Repo.update(repo_opts)
+    end
+
+    :ok
   end
 
   defp build_uploader(opts) do
@@ -188,41 +194,27 @@ defmodule BDS.Publishing do
   end
 
   defp run_command_upload(project_id, target, files, credentials, runner, ssh_auth_sock) do
-    Enum.reduce_while(files, :ok, fn relative_path, :ok ->
-      local_path = Path.join(target.local_dir, relative_path)
+    with {:ok, files_with_mtimes} <- collect_file_mtimes(target.local_dir, files) do
+      files_to_upload =
+        filter_scp_uploads(project_id, credentials, target.kind, files_with_mtimes)
 
-      with {:ok, local_mtime} <- file_mtime(local_path),
-           true <-
-             should_upload_scp_file?(
-               project_id,
-               credentials,
-               target.kind,
-               relative_path,
-               local_mtime
-             ) do
-        remote_path = remote_file_spec(credentials, target.remote_dir, relative_path)
+      case upload_scp_files(
+             project_id,
+             target,
+             credentials,
+             runner,
+             ssh_auth_sock,
+             files_to_upload,
+             []
+           ) do
+        {:ok, uploaded_files} ->
+          persist_uploaded_scp_files(project_id, credentials, target.kind, uploaded_files)
+          :ok
 
-        case run_command(runner, "scp", ["-q", local_path, remote_path], ssh_auth_sock) do
-          :ok ->
-            :ok =
-              mark_uploaded_scp_file(
-                project_id,
-                credentials,
-                target.kind,
-                relative_path,
-                local_mtime
-              )
-
-            {:cont, :ok}
-
-          {:error, reason} ->
-            {:halt, {:error, reason}}
-        end
-      else
-        false -> {:cont, :ok}
-        {:error, reason} -> {:halt, {:error, reason}}
+        {:error, reason} ->
+          {:error, reason}
       end
-    end)
+    end
   end
 
   defp run_command(runner, command, args, ssh_auth_sock) do
@@ -254,22 +246,98 @@ defmodule BDS.Publishing do
     end
   end
 
-  defp should_upload_scp_file?(project_id, credentials, target_kind, relative_path, local_mtime) do
+  defp filter_scp_uploads(project_id, credentials, target_kind, files_with_mtimes) do
     GenServer.call(
       __MODULE__,
-      {:should_upload_scp_file,
-       scp_upload_key(project_id, credentials, target_kind, relative_path), local_mtime}
+      {:filter_scp_uploads, project_id, credentials, target_kind, files_with_mtimes}
     )
   end
 
-  defp mark_uploaded_scp_file(project_id, credentials, target_kind, relative_path, local_mtime) do
+  defp record_uploaded_scp_files(project_id, credentials, target_kind, uploaded_files) do
     GenServer.call(
       __MODULE__,
-      {:mark_uploaded_scp_file,
-       scp_upload_key(project_id, credentials, target_kind, relative_path), local_mtime}
+      {:record_uploaded_scp_files, project_id, credentials, target_kind, uploaded_files}
     )
   end
 
+  defp upload_scp_files(
+         _project_id,
+         _target,
+         _credentials,
+         _runner,
+         _ssh_auth_sock,
+         [],
+         uploaded_files
+       ) do
+    {:ok, Enum.reverse(uploaded_files)}
+  end
+
+  defp upload_scp_files(
+         project_id,
+         target,
+         credentials,
+         runner,
+         ssh_auth_sock,
+         [{relative_path, local_mtime} | rest],
+         uploaded_files
+       ) do
+    local_path = Path.join(target.local_dir, relative_path)
+    remote_path = remote_file_spec(credentials, target.remote_dir, relative_path)
+
+    case run_command(runner, "scp", ["-q", local_path, remote_path], ssh_auth_sock) do
+      :ok ->
+        upload_scp_files(
+          project_id,
+          target,
+          credentials,
+          runner,
+          ssh_auth_sock,
+          rest,
+          [{relative_path, local_mtime} | uploaded_files]
+        )
+
+      {:error, reason} ->
+        persist_uploaded_scp_files(project_id, credentials, target.kind, uploaded_files)
+        {:error, reason}
+    end
+  end
+
+  defp collect_file_mtimes(local_dir, files) do
+    Enum.reduce_while(files, {:ok, []}, fn relative_path, {:ok, acc} ->
+      local_path = Path.join(local_dir, relative_path)
+
+      case file_mtime(local_path) do
+        {:ok, local_mtime} -> {:cont, {:ok, [{relative_path, local_mtime} | acc]}}
+        {:error, reason} -> {:halt, {:error, reason}}
+      end
+    end)
+    |> case do
+      {:ok, files_with_mtimes} -> {:ok, Enum.reverse(files_with_mtimes)}
+      {:error, reason} -> {:error, reason}
+    end
+  end
+
+  defp persist_uploaded_scp_files(_project_id, _credentials, _target_kind, []), do: :ok
+
+  defp persist_uploaded_scp_files(project_id, credentials, target_kind, uploaded_files) do
+    record_uploaded_scp_files(
+      project_id,
+      credentials,
+      target_kind,
+      Enum.reverse(uploaded_files)
+    )
+  end
+
+  defp should_upload_mtime?(nil, _local_mtime), do: true
+  defp should_upload_mtime?(recorded_mtime, local_mtime), do: local_mtime > recorded_mtime
+
+  defp repo_call_opts do
+    case Process.whereis(__MODULE__) do
+      pid when is_pid(pid) -> [caller: pid]
+      _other -> []
+    end
+  end
+
   defp scp_upload_key(project_id, credentials, target_kind, relative_path) do
     {
       project_id,

lib/bds/tasks.ex

@@ -7,11 +7,14 @@ defmodule BDS.Tasks do
   @default_progress_throttle_ms 250
   @default_recent_finished_limit 10
   @default_finished_task_ttl_ms :timer.hours(1)
+  @topic "tasks"
 
   def start_link(_opts) do
     GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
   end
 
+  def topic, do: @topic
+
   def submit_task(name, work, attrs \\ %{})
       when is_binary(name) and is_function(work, 1) and is_map(attrs) do
     GenServer.call(__MODULE__, {:submit_task, name, work, attrs})
@@ -66,9 +69,10 @@ defmodule BDS.Tasks do
     {:ok,
      %{
        tasks: %{},
-       queue: [],
+       queue: :queue.new(),
        running: %{},
-       ref_to_task: %{}
+       ref_to_task: %{},
+       finished_task_eviction_timer: nil
      }}
   end
 
@@ -80,8 +84,7 @@ defmodule BDS.Tasks do
     if map_size(next_state.running) < max_concurrent() do
       {:reply, {:ok, public_task(task)}, start_task(next_state, task.id, work)}
     else
-      {:reply, {:ok, public_task(task)},
-       %{next_state | queue: next_state.queue ++ [{task.id, work}]}}
+      {:reply, {:ok, public_task(task)}, enqueue_task(next_state, task.id, work)}
     end
   end
 
@@ -136,18 +139,20 @@ defmodule BDS.Tasks do
           |> start_queued_tasks()
           |> schedule_finished_task_eviction()
 
+        broadcast_terminal_task(next_state.tasks[task_id])
+
         {:reply, :ok, next_state}
 
-      Enum.any?(state.queue, fn {queued_id, _work} -> queued_id == task_id end) ->
+      queued_task?(state.queue, task_id) ->
         next_state =
           state
           |> update_task(task_id, %{status: :cancelled, finished_at: DateTime.utc_now()})
-          |> Map.update!(:queue, fn queue ->
-            Enum.reject(queue, fn {queued_id, _work} -> queued_id == task_id end)
-          end)
+          |> remove_queued_task(task_id)
           |> start_queued_tasks()
           |> schedule_finished_task_eviction()
 
+        broadcast_terminal_task(next_state.tasks[task_id])
+
         {:reply, :ok, next_state}
 
       state.tasks[task_id] == nil ->
@@ -179,6 +184,8 @@ defmodule BDS.Tasks do
       |> start_queued_tasks()
       |> schedule_finished_task_eviction()
 
+    broadcast_terminal_task(next_state.tasks[task_id])
+
     {:reply, :ok, next_state}
   end
 
@@ -193,6 +200,8 @@ defmodule BDS.Tasks do
       |> start_queued_tasks()
       |> schedule_finished_task_eviction()
 
+    broadcast_terminal_task(next_state.tasks[task_id])
+
     {:reply, :ok, next_state}
   end
 
@@ -202,7 +211,19 @@ defmodule BDS.Tasks do
   end
 
   def handle_info(:evict_finished_tasks, state) do
-    {:noreply, prune_expired_finished_tasks(state)}
+    next_state =
+      state
+      |> Map.put(:finished_task_eviction_timer, nil)
+      |> prune_expired_finished_tasks()
+
+    next_state =
+      if any_finished_tasks?(next_state) do
+        schedule_finished_task_eviction(next_state)
+      else
+        next_state
+      end
+
+    {:noreply, next_state}
   end
 
   def handle_info({ref, result}, state) do
@@ -240,6 +261,10 @@ defmodule BDS.Tasks do
           |> start_queued_tasks()
           |> schedule_finished_task_eviction()
 
+        if task.status != :cancelled do
+          broadcast_terminal_task(next_state.tasks[task_id])
+        end
+
         {:noreply, next_state}
     end
   end
@@ -271,6 +296,10 @@ defmodule BDS.Tasks do
           |> start_queued_tasks()
           |> schedule_finished_task_eviction()
 
+        if task.status != :cancelled and next_state.tasks[task_id].status == :failed do
+          broadcast_terminal_task(next_state.tasks[task_id])
+        end
+
         {:noreply, next_state}
     end
   end
@@ -296,11 +325,11 @@ defmodule BDS.Tasks do
       map_size(state.running) >= max_concurrent() ->
         state
 
-      state.queue == [] ->
+      :queue.is_empty(state.queue) ->
         state
 
       true ->
-        [{task_id, work} | remaining] = state.queue
+        {{:value, {task_id, work}}, remaining} = :queue.out(state.queue)
 
         state
         |> Map.put(:queue, remaining)
@@ -362,9 +391,21 @@ defmodule BDS.Tasks do
     end)
   end
 
+  defp broadcast_terminal_task(nil), do: :ok
+
+  defp broadcast_terminal_task(task) when task.status in [:completed, :failed, :cancelled] do
+    Phoenix.PubSub.broadcast(BDS.PubSub, topic(), {:task_terminal, public_task(task)})
+  end
+
+  defp broadcast_terminal_task(_task), do: :ok
+
   defp schedule_finished_task_eviction(state) do
-    Process.send_after(self(), :evict_finished_tasks, finished_task_ttl_ms())
-    state
+    if state.finished_task_eviction_timer do
+      state
+    else
+      timer_ref = Process.send_after(self(), :evict_finished_tasks, finished_task_ttl_ms())
+      %{state | finished_task_eviction_timer: timer_ref}
+    end
   end
 
   defp prune_expired_finished_tasks(state) do
@@ -378,6 +419,32 @@ defmodule BDS.Tasks do
     %{state | tasks: tasks}
   end
 
+  defp enqueue_task(state, task_id, work) do
+    %{state | queue: :queue.in({task_id, work}, state.queue)}
+  end
+
+  defp queued_task?(queue, task_id) do
+    queue
+    |> :queue.to_list()
+    |> Enum.any?(fn {queued_id, _work} -> queued_id == task_id end)
+  end
+
+  defp remove_queued_task(state, task_id) do
+    remaining_queue =
+      state.queue
+      |> :queue.to_list()
+      |> Enum.reject(fn {queued_id, _work} -> queued_id == task_id end)
+      |> :queue.from_list()
+
+    %{state | queue: remaining_queue}
+  end
+
+  defp any_finished_tasks?(state) do
+    Enum.any?(state.tasks, fn {_task_id, task} ->
+      task.status in [:completed, :failed, :cancelled]
+    end)
+  end
+
   defp expired_finished_task?(%{status: status, finished_at: %DateTime{} = finished_at}, now)
        when status in [:completed, :failed, :cancelled] do
     DateTime.diff(now, finished_at, :millisecond) >= finished_task_ttl_ms()

test/bds/ai/chat_streaming_test.exs

@@ -104,25 +104,19 @@ defmodule BDS.AI.ChatStreamingTest do
 
     server = start_supervised!({Bandit, plug: StreamingChatPlug, port: 0, startup_log: false})
     {:ok, {_address, port}} = ThousandIsland.listener_info(server)
-
-    assert {:ok, _endpoint} =
-             BDS.AI.put_endpoint(:online, %{
-               url: "http://127.0.0.1:#{port}/v1",
-               api_key: "sk-stream",
-               model: "stream-model"
-             })
-
-    assert :ok = BDS.AI.set_airplane_mode(false)
     assert {:ok, conversation} = BDS.AI.start_chat(%{model: "stream-model"})
 
-    {:ok, conversation: conversation}
+    {:ok, conversation: conversation, streaming_port: port}
   end
 
   test "incremental content events arrive before the final reply and persistence matches", %{
-    conversation: conversation
+    conversation: conversation,
+    streaming_port: port
   } do
     conversation_id = conversation.id
 
+    configure_streaming_runtime!(port)
+
     assert {:ok, reply} =
              BDS.AI.send_chat_message(conversation_id, "tell me a story",
                event_target: self()
@@ -142,11 +136,16 @@ defmodule BDS.AI.ChatStreamingTest do
     assert assistant_message.token_usage_output == 4
   end
 
-  test "cancel_chat mid-stream aborts the HTTP request", %{conversation: conversation} do
+  test "cancel_chat mid-stream aborts the HTTP request", %{
+    conversation: conversation,
+    streaming_port: port
+  } do
     Application.put_env(:bds, :chat_stream_scenario, :endless)
     conversation_id = conversation.id
     test_pid = self()
 
+    configure_streaming_runtime!(port)
+
     task =
       Task.async(fn ->
         BDS.AI.send_chat_message(conversation_id, "stream forever", event_target: test_pid)
@@ -161,4 +160,16 @@ defmodule BDS.AI.ChatStreamingTest do
     # The server notices the closed connection — the request was truly aborted.
     assert_receive :sse_client_disconnected, 2_000
   end
+
+  defp configure_streaming_runtime!(port) do
+    assert {:ok, _endpoint} =
+             BDS.AI.put_endpoint(:online, %{
+               url: "http://127.0.0.1:#{port}/v1",
+               api_key: "sk-stream",
+               model: "stream-model"
+             })
+
+    assert :ok = BDS.AI.put_model_preference(:chat, "stream-model")
+    assert :ok = BDS.AI.set_airplane_mode(false)
+  end
 end

test/bds/cli_sync_test.exs

@@ -76,6 +76,47 @@ defmodule BDS.CliSyncTest do
     assert is_integer(seen_notification.seen_at)
   end
 
+  test "watcher skips notification queries when sqlite data_version is unchanged" do
+    test_pid = self()
+    data_version = :erlang.make_ref()
+    :persistent_term.put(data_version, [7, 7])
+
+    data_version_reader = fn ->
+      [next | rest] = :persistent_term.get(data_version)
+      :persistent_term.put(data_version, rest)
+      next
+    end
+
+    notification_fetcher = fn ->
+      send(test_pid, :notifications_fetched)
+      {:ok, []}
+    end
+
+    pruner = fn ->
+      send(test_pid, :notifications_pruned)
+      {:ok, %{processed: 0, unprocessed: 0}}
+    end
+
+    on_exit(fn -> :persistent_term.erase(data_version) end)
+
+    watcher =
+      start_supervised!(
+        {Watcher,
+         poll_interval_ms: 60_000,
+         data_version_reader: data_version_reader,
+         notification_fetcher: notification_fetcher,
+         pruner: pruner}
+      )
+
+    :ok = Watcher.poll_now(watcher)
+    assert_receive :notifications_fetched, 500
+    assert_receive :notifications_pruned, 500
+
+    :ok = Watcher.poll_now(watcher)
+    refute_receive :notifications_fetched, 100
+    refute_receive :notifications_pruned, 100
+  end
+
   test "processed notifications are pruned after one hour and unprocessed notifications after one day" do
     now = BDS.Persistence.now_ms()
 

test/bds/csm020_nested_case_test.exs

@@ -76,18 +76,18 @@ defmodule BDS.CSM020NestedCaseTest do
     end
   end
 
-  describe "Publishing.handle_call :update_job uses with" do
+  describe "Publishing.update_job/2 uses with" do
     test "source code uses with instead of case" do
       source = File.read!("lib/bds/publishing.ex")
 
       [func_source] =
-        Regex.scan(~r/def handle_call\(\{:update_job.*?(?=\n  def |\n  @impl)/s, source)
+        Regex.scan(~r/defp update_job\(job_id, attrs\).*?(?=\n  defp |\nend)/s, source)
 
       assert func_source |> List.first() |> String.contains?("with"),
-             "update_job handler should use with"
+             "update_job should use with"
 
       refute func_source |> List.first() |> String.contains?("case Repo.get"),
-             "update_job handler should not use case Repo.get"
+             "update_job should not use case Repo.get"
     end
   end
 end

test/bds/csm036_impl_true_test.exs

@@ -14,7 +14,7 @@ defmodule BDS.CSM036ImplTrueTest do
           String.contains?(line, "def handle_call(")
         end)
 
-      assert length(handle_call_lines) >= 5, "expected at least 5 handle_call clauses"
+      assert length(handle_call_lines) >= 2, "expected at least 2 handle_call clauses"
 
       for {_line, idx} <- handle_call_lines do
         preceding = Enum.at(lines, idx - 2)

test/bds/desktop/shell_commands_test.exs

@@ -914,6 +914,21 @@ defmodule BDS.Desktop.ShellCommandsTest do
     assert message =~ "Project database is not initialized"
   end
 
+  test "rebuild sequencing waits on task messages instead of sleep polling" do
+    source = File.read!("lib/bds/desktop/shell_commands.ex")
+
+    func_source =
+      Regex.scan(~r/defp wait_for_group_phase(?:_message)?\(.*?(?=\n  defp |\nend)/s, source)
+      |> Enum.map(&List.first/1)
+      |> Enum.join("\n")
+
+    refute String.contains?(func_source, "Process.sleep"),
+           "wait_for_group_phase should not use sleep polling"
+
+    assert String.contains?(func_source, "Phoenix.PubSub.subscribe")
+    assert String.contains?(func_source, "receive")
+  end
+
   defp wait_for_task(task_id, matcher, timeout \\ 2_000)
 
   defp wait_for_task(task_id, _matcher, timeout) when timeout <= 0 do

test/bds/frontmatter_test.exs

@@ -0,0 +1,30 @@
+defmodule BDS.FrontmatterTest do
+  use ExUnit.Case, async: true
+
+  test "parse_document accepts CRLF frontmatter documents" do
+    contents = "---\r\ntitle: Hello\r\ntags:\r\n  - elixir\r\n---\r\nBody\r\n"
+
+    assert {:ok, %{fields: fields, body: body}} = BDS.Frontmatter.parse_document(contents)
+    assert fields["title"] == "Hello"
+    assert fields["tags"] == ["elixir"]
+    assert body == "Body"
+  end
+
+  test "serialize_document roundtrips quoted strings with embedded quotes and escapes" do
+    fields = [
+      {"title", "He said \"hi\" \\\\ there"},
+      {"summary", "Ends with a quote\""},
+      {"excerpt", "first line\nsecond line"}
+    ]
+
+    contents = BDS.Frontmatter.serialize_document(fields, "Body")
+
+    assert {:ok, %{fields: parsed_fields, body: body}} =
+             BDS.Frontmatter.parse_document(contents)
+
+    assert parsed_fields["title"] == "He said \"hi\" \\\\ there"
+    assert parsed_fields["summary"] == "Ends with a quote\""
+    assert parsed_fields["excerpt"] == "first line\nsecond line"
+    assert body == "Body"
+  end
+end

test/bds/publishing_test.exs

@@ -280,6 +280,46 @@ defmodule BDS.PublishingTest do
     assert elem(html_upload, 1) == ["-q", html_index, "deploy@example.com:/srv/blog/index.html"]
   end
 
+  test "upload_site batches scp mtime bookkeeping instead of calling the publishing server per file",
+       %{project: project, temp_dir: temp_dir} do
+    test_pid = self()
+
+    File.mkdir_p!(Path.join([temp_dir, "html", "posts"]))
+
+    for index <- 1..5 do
+      File.write!(Path.join([temp_dir, "html", "posts", "entry-#{index}.html"]), "<html />")
+    end
+
+    credentials = %{
+      ssh_host: "example.com",
+      ssh_user: "deploy",
+      ssh_remote_path: "/srv/blog",
+      ssh_mode: :scp
+    }
+
+    publishing_pid = Process.whereis(BDS.Publishing)
+    :erlang.trace(publishing_pid, true, [:receive])
+
+    runner = fn command, args, opts ->
+      send(test_pid, {:command_run, command, args, opts})
+      {"", 0}
+    end
+
+    assert {:ok, job} =
+             BDS.Publishing.upload_site(project.id, credentials,
+               command_runner: runner,
+               ssh_auth_sock: "/tmp/test-agent.sock"
+             )
+
+    assert wait_for_publish_job(job.id, &(&1.status == :completed)).status == :completed
+
+    :erlang.trace(publishing_pid, false, [:receive])
+
+    bookkeeping_calls = collect_publishing_bookkeeping_calls(publishing_pid)
+
+    assert length(bookkeeping_calls) <= 6
+  end
+
   test "publish jobs survive a publishing server restart because they are persisted", %{
     project: project,
     temp_dir: temp_dir
@@ -325,6 +365,25 @@ defmodule BDS.PublishingTest do
     end
   end
 
+  defp collect_publishing_bookkeeping_calls(publishing_pid, acc \\ []) do
+    receive do
+      {:trace, ^publishing_pid, :receive, {:"$gen_call", _from, message}}
+      when is_tuple(message) and tuple_size(message) > 0 and
+             elem(message, 0) in [
+               :should_upload_scp_file,
+               :mark_uploaded_scp_file,
+               :filter_scp_uploads,
+               :record_uploaded_scp_files
+             ] ->
+        collect_publishing_bookkeeping_calls(publishing_pid, [message | acc])
+
+      {:trace, ^publishing_pid, :receive, _message} ->
+        collect_publishing_bookkeeping_calls(publishing_pid, acc)
+    after
+      50 -> Enum.reverse(acc)
+    end
+  end
+
   defp wait_for_publish_job(job_id, predicate, attempts \\ 100)
 
   defp wait_for_publish_job(job_id, predicate, attempts) when attempts > 0 do

test/bds/search_test.exs

@@ -553,6 +553,15 @@ defmodule BDS.SearchTest do
     assert Enum.uniq(languages) == languages
   end
 
+  test "detect_language classifies umlaut-free German text as German" do
+    assert BDS.Search.detect_language("Der Fluss fliesst ruhig am Morgen entlang der alten Bruecke") ==
+             "de"
+  end
+
+  test "detect_language classifies accent-free French text as French" do
+    assert BDS.Search.detect_language("Je cours au parc chaque matin avant le travail") == "fr"
+  end
+
   test "search_posts finds translation text in multiple languages after reindex", %{
     project: project
   } do

test/bds/tasks_test.exs

@@ -266,6 +266,51 @@ defmodule BDS.TasksTest do
     assert running.id in task_ids
   end
 
+  test "finished task eviction uses a single live timer" do
+    Application.put_env(:bds, :tasks,
+      max_concurrent: 3,
+      progress_throttle_ms: 250,
+      finished_task_ttl_ms: 50
+    )
+
+    assert {:ok, first} = BDS.Tasks.register_external_task("first finished")
+    assert {:ok, second} = BDS.Tasks.register_external_task("second finished")
+
+    assert :ok = BDS.Tasks.complete_task(first.id)
+    first_timer = :sys.get_state(BDS.Tasks).finished_task_eviction_timer
+    assert is_reference(first_timer)
+    assert is_integer(Process.read_timer(first_timer))
+
+    assert :ok = BDS.Tasks.complete_task(second.id)
+    second_timer = :sys.get_state(BDS.Tasks).finished_task_eviction_timer
+
+    assert second_timer == first_timer
+    assert is_integer(Process.read_timer(second_timer))
+  end
+
+  test "task queue implementation avoids list append churn" do
+    source = File.read!("lib/bds/tasks.ex")
+
+    assert String.contains?(source, ":queue"), "tasks queue should use :queue"
+
+    refute String.contains?(source, "queue ++"),
+           "tasks queue should not append with ++"
+  end
+
+  test "terminal task states are broadcast on PubSub" do
+    Phoenix.PubSub.subscribe(BDS.PubSub, BDS.Tasks.topic())
+
+    assert {:ok, completed} =
+             BDS.Tasks.submit_task("broadcast completion", fn _report -> {:ok, :done} end,
+               %{group_id: "broadcast-group", group_name: "Maintenance"}
+             )
+
+    assert_receive {:task_terminal, task_event}, 1_000
+    assert task_event.id == completed.id
+    assert task_event.group_id == "broadcast-group"
+    assert task_event.status == :completed
+  end
+
   defp receive_started do
     receive do
       {:started, name, pid} -> {name, pid}
@@ -290,4 +335,5 @@ defmodule BDS.TasksTest do
   defp wait_for_task(_task_id, _predicate, 0) do
     flunk("task did not reach expected state")
   end
+
 end