fix(worker): a source deleted before processing must fail permanently, not starve the queue (#1364)

* fix(worker): fail permanently when the source was deleted before processing

Source.get() raises NotFoundError instead of returning None, so the
'if not source: raise ValueError' guard in process_source_command was
unreachable. NotFoundError is not in the command's stop_on list, so the job
was retried as transient (max_attempts=15, exponential_jitter, wait_max=120)
and every job behind it starved until the worker process was restarted.

Catch NotFoundError around that call and re-raise as ValueError, which is
already this command's permanent-failure channel.

Fixes #1363

* chore(worker): rebase on main, cover deleted transformations, changelog

- Rebased onto main (import conflict with #1425).
- A transformation deleted before the job runs hits the same NotFoundError
  path in process_source_command; it now fails permanently too (test added).
- ruff format on the new test (CI gates formatting since #1429).
- CHANGELOG entry.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GBYpxzSS62Y3TZkgB9NUZy

* fix(domain): only a missing record is NotFoundError in ObjectModel.get

ObjectModel.get wrapped every exception from repo_query, including the
RuntimeError it re-raises for retriable SurrealDB transaction conflicts, as
NotFoundError. With process_source now treating NotFoundError as permanent,
a transient conflict would have failed the job for good. get() now lets
NotFoundError / InvalidInputError through and raises DatabaseOperationError
(chained) for anything else.

The permanent-failure messages are static; the underlying error stays as the
exception's cause for the worker log.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GBYpxzSS62Y3TZkgB9NUZy

* fix(worker): run_transformation fails permanently on a deleted source too

NotFoundError (now only raised for a genuinely missing record) joins
run_transformation's stop_on and permanent-failure arm.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GBYpxzSS62Y3TZkgB9NUZy

* docs(changelog): rebase entry onto the consolidated Unreleased section

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GBYpxzSS62Y3TZkgB9NUZy

* fix(worker): a source deleted mid-processing also fails permanently

save_source re-reads the source after extraction; a deletion in that window
raised NotFoundError from inside the graph, which process_source retried as
transient. NotFoundError (now only a genuinely missing record) joins its
stop_on and gets a permanent-failure arm.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GBYpxzSS62Y3TZkgB9NUZy

* docs(adr): ADR-013 for the ObjectModel.get error contract

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GBYpxzSS62Y3TZkgB9NUZy

* fix(embeddings): a record deleted before embedding is a permanent failure

Completes the review on #1364: embed_* commands treat NotFoundError (now only
a genuinely missing record) like their other permanent failures instead of
retrying it; a DatabaseOperationError still retries.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GBYpxzSS62Y3TZkgB9NUZy

* docs(embeddings): NotFoundError is a permanent failure too

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GBYpxzSS62Y3TZkgB9NUZy

---------

Co-authored-by: Luis Novo <lfnovo@gmail.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

CHANGELOG.md

@@ -25,6 +25,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **Credential encryption now uses PBKDF2-HMAC-SHA256 (600k iterations) with versioned `pbkdf2v1:` ciphertext.** New keys are written in the new format; legacy values keep decrypting, and a one-shot `POST /api/credentials/migrate-encryption` pass upgrades stored keys (idempotent, fail-closed per record). Downgrade warning: versions predating this change cannot read the new format — back up the database before migrating (#1317)
 
 ### Fixed
+- **The worker no longer stalls when a source is deleted before it is processed.** `Source.get()` raises `NotFoundError` for a missing record, so the job was retried as a transient failure (up to 15 attempts with backoff) and every job behind it waited; a missing source or transformation now fails the job immediately. Relatedly, loading a record no longer reports a database failure (e.g. a SurrealDB transaction conflict) as "not found": it raises a database error, so it stays retryable in jobs and returns 500 instead of 404 in the API (#1363)
 - **Model calls default to a 180-second timeout.** esperanto 2.28 now enforces `ESPERANTO_LLM_TIMEOUT` on every provider (60 s by default); before, most providers used their SDK default and Ollama waited indefinitely. Open Notebook now sets 180 s when the variable is unset, so long answers and slower models aren't cut off at 60 s. **Ollama and other slow local setups** that relied on unlimited waits should set `ESPERANTO_LLM_TIMEOUT` (and `API_CLIENT_TIMEOUT` above it) explicitly (#1434)
 - Batch large embedding inserts into groups of 50 records and reuse one database connection across batches to reduce oversized WebSocket payloads and repeated sign-ins.
 - `GET /api/notebooks` now normalizes `order_by` through the shared `_validate_order_by()` guard, keeping its stricter `name`/`created`/`updated` allowlist on top, so the two ORDER BY validators can no longer drift (#1416)

commands/embedding_commands.py

@@ -16,13 +16,17 @@ from surreal_commands import CommandInput, CommandOutput, command, submit_comman
 from open_notebook.ai.models import model_manager
 from open_notebook.database.repository import ensure_record_id, repo_insert, repo_query
 from open_notebook.domain.notebook import Note, Source, SourceInsight
-from open_notebook.exceptions import ConfigurationError, ContextLengthExceededError
+from open_notebook.exceptions import (
+    ConfigurationError,
+    ContextLengthExceededError,
+    NotFoundError,
+)
 from open_notebook.utils.chunking import ContentType, chunk_text, detect_content_type
 from open_notebook.utils.embedding import generate_embedding, generate_embeddings
 
 # NOTE: `stop_on` below can never trigger in practice — each command catches
-# ValueError internally and returns success=False instead of raising, so the
-# retry layer never sees it. Kept as-is on purpose; to be revisited in a
+# ValueError (and NotFoundError) internally and returns success=False instead
+# of raising, so the retry layer never sees it. Kept as-is on purpose; to be revisited in a
 # dedicated error-handling PR.
 EMBED_RETRY_CONFIG = {
     "max_attempts": 5,
@@ -66,7 +70,8 @@ async def _embed_record(
     Returns:
         (extra_output_fields, processing_time, error_message)
         extra_output_fields is None and error_message is set on permanent
-        (ValueError) failure. Transient failures re-raise so the retry layer
+        (ValueError, or NotFoundError for a record deleted before the job ran)
+        failure. Transient failures re-raise so the retry layer
         can handle them.
     """
     start_time = time.time()
@@ -82,8 +87,10 @@ async def _embed_record(
         )
         return extra_fields, processing_time, None
 
-    except ValueError as e:
-        # Permanent failure - don't retry
+    except (ValueError, NotFoundError) as e:
+        # Permanent failure - don't retry. NotFoundError means the record was
+        # deleted before the job ran (ObjectModel.get raises it only for a
+        # missing record; DB failures are DatabaseOperationError and retry).
         processing_time = time.time() - start_time
         cmd_id = get_command_id(input_data)
         logger.error(f"Failed to embed {kind} {record_id} (command: {cmd_id}): {e}")
@@ -233,7 +240,8 @@ async def embed_note_command(input_data: EmbedNoteInput) -> EmbedNoteOutput:
     Retry Strategy:
     - Retries up to 5 times for transient failures (network, timeout, etc.)
     - Uses exponential-jitter backoff (1-60s)
-    - Does NOT retry permanent failures (ValueError for validation errors)
+    - Does NOT retry permanent failures (ValueError for validation errors,
+      NotFoundError for a record deleted before the job ran)
     """
 
     async def embed() -> Tuple[Dict[str, Any], str]:
@@ -275,7 +283,8 @@ async def embed_insight_command(input_data: EmbedInsightInput) -> EmbedInsightOu
     Retry Strategy:
     - Retries up to 5 times for transient failures (network, timeout, etc.)
     - Uses exponential-jitter backoff (1-60s)
-    - Does NOT retry permanent failures (ValueError for validation errors)
+    - Does NOT retry permanent failures (ValueError for validation errors,
+      NotFoundError for a record deleted before the job ran)
     """
 
     async def embed() -> Tuple[Dict[str, Any], str]:
@@ -320,7 +329,8 @@ async def embed_source_command(input_data: EmbedSourceInput) -> EmbedSourceOutpu
     Retry Strategy:
     - Retries up to 5 times for transient failures (network, timeout, etc.)
     - Uses exponential-jitter backoff (1-60s)
-    - Does NOT retry permanent failures (ValueError for validation errors)
+    - Does NOT retry permanent failures (ValueError for validation errors,
+      NotFoundError for a record deleted before the job ran)
     """
 
     async def embed() -> Tuple[Dict[str, Any], str]:
@@ -440,7 +450,8 @@ async def create_insight_command(
     Retry Strategy:
     - Retries up to 5 times for transient failures (network, timeout, etc.)
     - Uses exponential-jitter backoff (1-60s)
-    - Does NOT retry permanent failures (ValueError for validation errors)
+    - Does NOT retry permanent failures (ValueError for validation errors,
+      NotFoundError for a record deleted before the job ran)
     """
     start_time = time.time()
 

commands/source_commands.py

@@ -13,6 +13,7 @@ from open_notebook.exceptions import (
     ContextLengthExceededError,
     IncompleteGenerationError,
     InvalidInputError,
+    NotFoundError,
 )
 
 try:
@@ -53,7 +54,10 @@ class SourceProcessingOutput(CommandOutput):
             ConfigurationError,
             ContextLengthExceededError,
             IncompleteGenerationError,
-        ],  # Don't retry validation/config errors or incomplete generations
+            NotFoundError,
+        ],  # Don't retry validation/config errors, incomplete generations, or a
+        # record deleted mid-processing (NotFoundError means "missing", never a
+        # DB failure: ObjectModel.get raises DatabaseOperationError for those)
         "retry_log_level": "debug",  # Avoid log noise during transaction conflicts
     },
 )
@@ -75,7 +79,11 @@ async def process_source_command(
         transformations = []
         for trans_id in input_data.transformations:
             logger.info(f"Loading transformation: {trans_id}")
-            transformation = await Transformation.get(trans_id)
+            try:
+                transformation = await Transformation.get(trans_id)
+            except NotFoundError as e:
+                # Same as a deleted source below: permanent, not transient.
+                raise ValueError(f"Transformation '{trans_id}' no longer exists") from e
             if not transformation:
                 raise ValueError(f"Transformation '{trans_id}' not found")
             transformations.append(transformation)
@@ -83,7 +91,18 @@ async def process_source_command(
         logger.info(f"Loaded {len(transformations)} transformations")
 
         # 2. Get existing source record to update its command field
-        source = await Source.get(input_data.source_id)
+        try:
+            source = await Source.get(input_data.source_id)
+        except NotFoundError as e:
+            # The source was removed after this job was queued (e.g. a sync tool
+            # deleted it, or the record was cleaned up). The job can never
+            # succeed: raise a permanent error (ValueError is in `stop_on`) so
+            # surreal-commands marks it failed instead of spending 15 retries
+            # with exponential backoff, which starves every job behind it.
+            raise ValueError(
+                f"Source '{input_data.source_id}' no longer exists "
+                "(deleted before processing?)"
+            ) from e
         if not source:
             raise ValueError(f"Source '{input_data.source_id}' not found")
 
@@ -142,6 +161,13 @@ async def process_source_command(
             f"Generation failed (permanent) for source {input_data.source_id}: {e}"
         )
         raise  # Preserve failed job status; stop_on prevents automatic retries.
+    except NotFoundError as e:
+        # E.g. the source was deleted while extraction ran (save_source re-reads
+        # it). Permanent: stop_on prevents retries that would starve the queue.
+        logger.error(
+            f"Source processing failed (permanent), record no longer exists: {e}"
+        )
+        raise
     except ValueError as e:
         # Validation errors are permanent failures. Re-raise so surreal-commands
         # marks the job as `failed` (stop_on=[ValueError] already prevents
@@ -193,7 +219,9 @@ class RunTransformationOutput(CommandOutput):
             ContextLengthExceededError,
             IncompleteGenerationError,
             InvalidInputError,
-        ],  # Don't retry validation/config errors or incomplete generations
+            NotFoundError,
+        ],  # Don't retry validation/config errors, incomplete generations, or
+        # a source/transformation deleted before the job ran
         "retry_log_level": "warning",
     },
 )
@@ -257,7 +285,7 @@ async def run_transformation_command(
             processing_time=processing_time,
         )
 
-    except (IncompleteGenerationError, InvalidInputError) as e:
+    except (IncompleteGenerationError, InvalidInputError, NotFoundError) as e:
         # e.g. the source has no text to transform
         logger.error(
             f"Generation failed (permanent) for transformation "

docs/7-DEVELOPMENT/decisions/ADR-013-objectmodel-get-error-contract.md

@@ -0,0 +1,30 @@
+# ADR-013: ObjectModel.get raises NotFoundError only for a missing record
+
+- **Status**: Accepted
+- **Date**: 2026-10
+- **Related**: #1363, PR #1364
+
+## Context
+
+`ObjectModel.get()` is the shared fetch boundary for every domain model. It wrapped **every** exception from `repo_query` as `NotFoundError("Object with id ... not found - {e}")`, including connection errors and the `RuntimeError` that `repo_query` re-raises for retriable SurrealDB v2 transaction conflicts. Callers treat `NotFoundError` as permanent: routers map it to 404, and since #1363 the background commands treat it as a non-retryable job failure. Under the old wrapping, a transient database hiccup would have been reported as "not found" and never retried.
+
+## Decision
+
+**`ObjectModel.get()` raises `NotFoundError` only when the record does not exist.** `InvalidInputError` (empty id, unknown table) passes through unchanged. Any other failure is raised as `DatabaseOperationError`, chained to the original exception.
+
+Consequences for callers:
+
+- **Commands:** `NotFoundError` means "the record was deleted" and is permanent. `process_source` and `run_transformation` have it in `stop_on`. `DatabaseOperationError` stays retryable.
+- **API:** a missing record is still 404. A database failure during a fetch becomes a 500 (`OpenNotebookError` handler) instead of a misleading 404.
+
+## Alternatives considered
+
+- **Keep the wrapping and catch `NotFoundError` per command**: rejected, because it makes every transient conflict permanent, exactly the failure the worker's retries exist for.
+- **Match the error message to tell the two apart**: rejected as fragile string coupling.
+- **Return `None` for a missing record**: rejected. It's a larger API change, and the `if not x` guards would have to come back everywhere.
+
+## Consequences
+
+- A command can safely treat `NotFoundError` as permanent (deleted before or during processing) without starving the queue on DB conflicts.
+- API clients see 500, not 404, for database outages on single-record fetches. That's correct, but it's visible if a client relied on the old 404.
+- New fetch helpers should follow the same split: "missing" vs "failed".

docs/7-DEVELOPMENT/decisions/README.md

@@ -53,5 +53,6 @@ What this makes easier, what it makes harder, what to watch. (bullets)
 | [ADR-009](ADR-009-pbkdf2-credential-encryption.md) | PBKDF2 credential key derivation with versioned ciphertext | Accepted |
 | [ADR-011](ADR-011-design-token-system.md) | Visual identity is a token contract in globals.css, reviewed through /dev/design | Accepted |
 | [ADR-012](ADR-012-provider-endpoint-overrides.md) | Provider endpoint overrides are declared in the registry | Accepted |
+| [ADR-013](ADR-013-objectmodel-get-error-contract.md) | ObjectModel.get raises NotFoundError only for a missing record | Accepted |
 | [PDR-001](PDR-001-single-user-first.md) | Single-user first; don't preclude multi-user | Accepted |
 | [PDR-002](PDR-002-provider-agnostic-core.md) | Provider-agnostic core by default | Accepted |

open_notebook/domain/base.py

@@ -125,10 +125,15 @@ class ObjectModel(BaseModel):
                 return target_class(**result[0])
             else:
                 raise NotFoundError(f"{table_name} with id {id} not found")
+        except (NotFoundError, InvalidInputError):
+            raise
         except Exception as e:
+            # A database failure (connection, SurrealDB transaction conflict...)
+            # is not "not found": callers treat NotFoundError as permanent (404,
+            # no retry), so surface it as DatabaseOperationError instead.
             logger.error(f"Error fetching object with id {id}: {str(e)}")
             logger.exception(e)
-            raise NotFoundError(f"Object with id {id} not found - {str(e)}")
+            raise DatabaseOperationError(f"Failed to fetch object with id {id}") from e
 
     @classmethod
     def _get_class_by_table_name(cls, table_name: str) -> Optional[Type["ObjectModel"]]:

tests/test_source_deleted_before_processing.py

@@ -0,0 +1,204 @@
+"""A source deleted before its process_source job runs must fail permanently.
+
+Regression for the worker-starvation bug: `Source.get()` raises `NotFoundError`
+instead of returning `None`, so the `if not source: raise ValueError(...)` guard in
+`process_source_command` never fired. `NotFoundError` is not in the command's
+`stop_on` list, so the job was retried as a transient failure (up to 15 attempts
+with exponential backoff, waits up to 120s) and every job behind it starved —
+the worker stopped consuming jobs until the process was restarted.
+"""
+
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+import commands  # noqa: F401 -- import registers the @command decorators
+from open_notebook.exceptions import NotFoundError
+
+
+@pytest.mark.asyncio
+async def test_missing_source_raises_permanent_error():
+    """NotFoundError from Source.get must surface as ValueError (terminal)."""
+    from commands.source_commands import SourceProcessingInput, process_source_command
+
+    input_data = SourceProcessingInput(
+        source_id="source:does-not-exist",
+        content_state={"file_path": "/tmp/whatever.md"},
+        notebook_ids=["notebook:whatever"],
+        transformations=[],
+        embed=True,
+    )
+
+    with patch(
+        "commands.source_commands.Source.get",
+        new=AsyncMock(
+            side_effect=NotFoundError("source with id source:does-not-exist not found")
+        ),
+    ):
+        with pytest.raises(ValueError) as excinfo:
+            await process_source_command(input_data)
+
+    assert "no longer exists" in str(excinfo.value)
+    # The retry policy keys off the exception type: ValueError is in `stop_on`,
+    # NotFoundError is not — that difference is the whole fix.
+    assert not isinstance(excinfo.value, NotFoundError)
+
+
+@pytest.mark.asyncio
+async def test_missing_transformation_raises_permanent_error():
+    """A transformation deleted before the job runs is just as permanent."""
+    from commands.source_commands import SourceProcessingInput, process_source_command
+
+    input_data = SourceProcessingInput(
+        source_id="source:abc",
+        content_state={"file_path": "/tmp/whatever.md"},
+        notebook_ids=["notebook:whatever"],
+        transformations=["transformation:gone"],
+        embed=True,
+    )
+
+    with patch(
+        "commands.source_commands.Transformation.get",
+        new=AsyncMock(side_effect=NotFoundError("transformation not found")),
+    ):
+        with pytest.raises(ValueError, match="no longer exists"):
+            await process_source_command(input_data)
+
+
+@pytest.mark.asyncio
+async def test_object_get_reports_db_failures_as_database_errors():
+    """ObjectModel.get used to wrap every exception (e.g. a retriable SurrealDB
+    transaction conflict) as NotFoundError, so the fix above would have made
+    transient failures permanent. Only a missing record is NotFoundError."""
+    from open_notebook.domain.notebook import Source
+    from open_notebook.exceptions import DatabaseOperationError
+
+    with patch(
+        "open_notebook.domain.base.repo_query",
+        new=AsyncMock(side_effect=RuntimeError("Transaction conflict: retry")),
+    ):
+        with pytest.raises(DatabaseOperationError):
+            await Source.get("source:abc")
+
+    with patch("open_notebook.domain.base.repo_query", new=AsyncMock(return_value=[])):
+        with pytest.raises(NotFoundError):
+            await Source.get("source:abc")
+
+
+@pytest.mark.asyncio
+async def test_transient_db_failure_stays_retryable():
+    """A database failure while loading the source is re-raised as is (not
+    ValueError), so the worker's retry policy still applies."""
+    from commands.source_commands import SourceProcessingInput, process_source_command
+    from open_notebook.exceptions import DatabaseOperationError
+
+    input_data = SourceProcessingInput(
+        source_id="source:abc",
+        content_state={"file_path": "/tmp/whatever.md"},
+        notebook_ids=["notebook:whatever"],
+        transformations=[],
+        embed=True,
+    )
+
+    with patch(
+        "commands.source_commands.Source.get",
+        new=AsyncMock(side_effect=DatabaseOperationError("Failed to fetch")),
+    ):
+        with pytest.raises(DatabaseOperationError):
+            await process_source_command(input_data)
+
+
+@pytest.mark.asyncio
+async def test_run_transformation_fails_permanently_when_source_is_gone():
+    """run_transformation had the same deletion-before-processing gap."""
+    from surreal_commands import registry
+
+    from commands.source_commands import (
+        RunTransformationInput,
+        run_transformation_command,
+    )
+
+    with patch(
+        "commands.source_commands.Source.get",
+        new=AsyncMock(side_effect=NotFoundError("source not found")),
+    ):
+        with pytest.raises(NotFoundError):
+            await run_transformation_command(
+                RunTransformationInput(
+                    source_id="source:gone", transformation_id="transformation:t1"
+                )
+            )
+
+    command = registry.get_command("open_notebook", "run_transformation")
+    assert command is not None
+    assert NotFoundError in command.retry_config.stop_on
+
+
+@pytest.mark.asyncio
+async def test_source_deleted_during_processing_fails_permanently():
+    """save_source re-reads the source after extraction; a deletion in that
+    window raises NotFoundError from inside the graph, which must not be
+    retried either."""
+    from types import SimpleNamespace
+
+    from surreal_commands import registry
+
+    from commands.source_commands import SourceProcessingInput, process_source_command
+
+    source = SimpleNamespace(id="source:abc", command=None, save=AsyncMock())
+    with (
+        patch(
+            "commands.source_commands.Source.get", new=AsyncMock(return_value=source)
+        ),
+        patch(
+            "commands.source_commands.source_graph.ainvoke",
+            new=AsyncMock(
+                side_effect=NotFoundError("source with id source:abc not found")
+            ),
+        ),
+    ):
+        with pytest.raises(NotFoundError):
+            await process_source_command(
+                SourceProcessingInput(
+                    source_id="source:abc",
+                    content_state={"file_path": "/tmp/x.md"},
+                    notebook_ids=[],
+                    transformations=[],
+                    embed=False,
+                )
+            )
+
+    command = registry.get_command("open_notebook", "process_source")
+    assert command is not None
+    assert NotFoundError in command.retry_config.stop_on
+
+
+@pytest.mark.asyncio
+async def test_embedding_a_deleted_source_fails_permanently():
+    """embed_source had the same gap (lfnovo's review on #1364): a record
+    deleted before the job ran is a permanent failure, not a retry."""
+    from commands.embedding_commands import EmbedSourceInput, embed_source_command
+
+    with patch(
+        "commands.embedding_commands.Source.get",
+        new=AsyncMock(
+            side_effect=NotFoundError("source with id source:gone not found")
+        ),
+    ):
+        result = await embed_source_command(EmbedSourceInput(source_id="source:gone"))
+
+    assert result.success is False
+    assert "not found" in (result.error_message or "")
+
+
+@pytest.mark.asyncio
+async def test_embedding_db_failure_stays_retryable():
+    from commands.embedding_commands import EmbedSourceInput, embed_source_command
+    from open_notebook.exceptions import DatabaseOperationError
+
+    with patch(
+        "commands.embedding_commands.Source.get",
+        new=AsyncMock(side_effect=DatabaseOperationError("Failed to fetch")),
+    ):
+        with pytest.raises(DatabaseOperationError):
+            await embed_source_command(EmbedSourceInput(source_id="source:abc"))