feat: introduced deliver queue

2026-08-26 09:34:58 +02:00
parent 41238c4125
commit 247180db63
38 changed files with 1699 additions and 904 deletions
@@ -25,6 +25,7 @@ BASANGO_CRAWLER_UPDATE_DIRECTION=forward
 BASANGO_CRAWLER_REDIS_URL=redis://localhost:6379/0
 BASANGO_CRAWLER_QUEUE_DISCOVERY=discovery
 BASANGO_CRAWLER_QUEUE_ARTICLES=articles
+BASANGO_CRAWLER_QUEUE_DELIVERY=delivery
 BASANGO_CRAWLER_RETAIN_COMPLETED=3600
 BASANGO_CRAWLER_RETAIN_FAILED=86400

@@ -39,3 +39,6 @@ jobs:

      - name: Run tests
        run: cargo test --all-targets --locked
+
+      - name: Run Clippy
+        run: cargo clippy --all-targets --all-features --locked -- -D warnings
@@ -48,7 +48,7 @@ sudo -u basango ./crawler schedule --source-id 7sur7.cd
 sudo -u basango ./crawler schedule --source-id 7sur7.cd --category sport --direction backward
 ```

-The worker consumes the queued jobs. Systemd is not used for scheduling.
+The worker runs discovery, article parsing, and API delivery concurrently. Articles are persisted to SQLite before a delivery job is queued, so a restart can safely resume unfinished delivery. Systemd is not used for scheduling.

 ### Direct crawl

@@ -69,7 +69,7 @@ cd /opt/crawler
 sudo -u basango ./crawler status
 ```

-Retry pending article deliveries:
+Manually retry pending article deliveries (normally the delivery queue handles these):

 ```bash
 cd /opt/crawler
@@ -3,6 +3,7 @@
    "prefix": "basango:crawler",
    "queues": {
      "articles": "articles",
+      "delivery": "delivery",
      "discovery": "discovery"
    },
    "redis_url": "redis://localhost:6379/0",
@@ -35,7 +35,7 @@ You can also omit the argument and answer the URL prompt, or set `BASANGO_CRAWLE

 Pushing a `v*` Git tag runs the release workflow, which publishes native `aarch64` (Raspberry Pi) and `x86_64` Linux archives to the GitHub release.

-Each agent ID prefixes its BullMQ queue names, so multiple Pis can safely share Redis. The installer does not schedule crawls. Run `crawler schedule` yourself or configure cron later with the sources and cadence assigned to that device.
+Each agent ID prefixes its discovery, article, and delivery queue names, so multiple Pis can safely share Redis. The worker consumes all three concurrently and reconciles SQLite delivery records after a restart. The installer does not schedule crawls. Run `crawler schedule` yourself or configure cron later with the sources and cadence assigned to that device.

 To reset a Pi, stop its worker before clearing its scoped queues and SQLite outbox:

@@ -12,7 +12,7 @@ mod outbox;
 pub(crate) use forwarder::endpoint_url;
 pub use forwarder::{ArticleIngestionClient, DeliveryResult};
 pub use normalize::normalize;
-pub use outbox::{DeliveryStatus, Outbox, OutboxEntry, OutboxStats};
+pub use outbox::{DeliveryIntent, DeliveryStatus, Outbox, OutboxEntry, OutboxStats};

 use crate::{
    domain::{Article, ArticleDraft},
@@ -3,7 +3,7 @@
 use std::collections::HashSet;

 use crate::{
-    domain::{Article, ArticleDraft},
+    domain::{Article, ArticleDraft, ArticleHash},
    error::{CrawlError, Result},
 };

@@ -19,7 +19,7 @@ pub fn normalize(draft: ArticleDraft) -> Result<Article> {
    }
    // The URL hash is an identity value, not a security boundary. It lets
    // SQLite enforce idempotency when a URL is crawled more than once.
-    let hash = format!("{:x}", md5::compute(draft.link.as_str()));
+    let hash = ArticleHash::from_url(&draft.link);

    let mut seen = HashSet::new();
    let categories = draft
@@ -4,67 +4,25 @@
 //! is held across `.await`. Clones share one connection inside a process; WAL
 //! mode still allows other crawler processes to coexist safely.

+mod intents;
+mod model;
+mod persistence;
+
 use std::{
    path::Path,
-    str::FromStr,
    sync::{Arc, Mutex, MutexGuard},
 };

 use chrono::{DateTime, Duration, Utc};
-use rusqlite::{
-    Connection, OptionalExtension, TransactionBehavior, named_params, params, types::Type,
-};
+use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params, types::Type};

 use crate::{
    domain::Article,
    error::{CrawlError, Result},
 };

-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum DeliveryStatus {
-    Pending,
-    Forwarded,
-    Failed,
-}
-
-impl FromStr for DeliveryStatus {
-    type Err = CrawlError;
-
-    fn from_str(value: &str) -> Result<Self> {
-        match value {
-            "pending" => Ok(Self::Pending),
-            "forwarded" => Ok(Self::Forwarded),
-            "failed" => Ok(Self::Failed),
-            other => Err(CrawlError::Configuration(format!(
-                "unknown outbox status '{other}'"
-            ))),
-        }
-    }
-}
-
-#[derive(Debug, Clone)]
-pub struct OutboxEntry {
-    pub article: Article,
-    pub status: DeliveryStatus,
-    pub attempts: u32,
-    pub retryable: bool,
-    pub last_error: Option<String>,
-    pub created_at: DateTime<Utc>,
-    pub updated_at: DateTime<Utc>,
-    pub forwarded_at: Option<DateTime<Utc>>,
-    pub claimed_at: Option<DateTime<Utc>>,
-    pub claimed_by: Option<String>,
-}
-
-#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
-pub struct OutboxStats {
-    pub total: usize,
-    pub pending: usize,
-    pub forwarded: usize,
-    pub failed: usize,
-    pub retryable_failed: usize,
-    pub claimed: usize,
-}
+pub use model::{DeliveryIntent, DeliveryStatus, OutboxEntry, OutboxStats};
+use persistence::save_article;

 #[derive(Clone)]
 pub struct Outbox {
@@ -113,7 +71,9 @@ impl Outbox {
                    COALESCE(SUM(status = 'forwarded'), 0),
                    COALESCE(SUM(status = 'failed'), 0),
                    COALESCE(SUM(status = 'failed' AND retryable = 1), 0),
-                    COALESCE(SUM(claimed_at IS NOT NULL), 0)
+                    COALESCE(SUM(claimed_at IS NOT NULL), 0),
+                    (SELECT COALESCE(SUM(status = 'pending'), 0) FROM delivery_intents),
+                    (SELECT COALESCE(SUM(status = 'failed'), 0) FROM delivery_intents)
                FROM articles"#,
                [],
                |row| {
@@ -124,6 +84,8 @@ impl Outbox {
                        failed: row.get(3)?,
                        retryable_failed: row.get(4)?,
                        claimed: row.get(5)?,
+                        delivery_intents_pending: row.get(6)?,
+                        delivery_intents_failed: row.get(7)?,
                    })
                },
            )
@@ -133,64 +95,8 @@ impl Outbox {
    /// Upsert by hash while preserving an already-forwarded state. This is the
    /// idempotency boundary: crawling the same URL twice does not redeliver it.
    pub fn save(&self, article: &Article) -> Result<DeliveryStatus> {
-        let timestamp = Utc::now().to_rfc3339();
-        let categories = serde_json::to_string(&article.categories)?;
-        let metadata = article
-            .metadata
-            .as_ref()
-            .map(serde_json::to_string)
-            .transpose()?;
-        let payload = serde_json::to_string(article)?;
-
        let connection = self.connection()?;
-        connection.execute(
-            r#"
-            INSERT INTO articles (
-                hash, source_id, link, title, body, categories, metadata,
-                published_at, payload, status, attempts, retryable,
-                last_error, created_at, updated_at, forwarded_at
-            )
-            VALUES (
-                :hash, :source_id, :link, :title, :body, :categories, :metadata,
-                :published_at, :payload, 'pending', 0, 1, NULL, :now, :now, NULL
-            )
-            ON CONFLICT(hash) DO UPDATE SET
-                source_id = excluded.source_id,
-                link = excluded.link,
-                title = excluded.title,
-                body = excluded.body,
-                categories = excluded.categories,
-                metadata = excluded.metadata,
-                published_at = excluded.published_at,
-                payload = excluded.payload,
-                status = CASE WHEN articles.status = 'forwarded' THEN 'forwarded' ELSE 'pending' END,
-                last_error = CASE WHEN articles.status = 'forwarded' THEN articles.last_error ELSE NULL END,
-                retryable = CASE WHEN articles.status = 'forwarded' THEN articles.retryable ELSE 1 END,
-                updated_at = excluded.updated_at,
-                forwarded_at = CASE WHEN articles.status = 'forwarded' THEN articles.forwarded_at ELSE NULL END,
-                claimed_at = NULL,
-                claimed_by = NULL
-            "#,
-            named_params! {
-                ":hash": article.hash,
-                ":source_id": article.source_id.as_str(),
-                ":link": article.link.as_str(),
-                ":title": article.title,
-                ":body": article.body,
-                ":categories": categories,
-                ":metadata": metadata,
-                ":published_at": article.published_at.to_rfc3339(),
-                ":payload": payload,
-                ":now": timestamp,
-            },
-        )?;
-
-        let status: String = connection.query_row(
-            "SELECT status FROM articles WHERE hash = ?1",
-            [&article.hash],
-            |row| row.get(0),
-        )?;
-        status.parse()
+        save_article(&connection, article)
    }

    pub fn list_pending(&self, source_id: Option<&str>, limit: usize) -> Result<Vec<OutboxEntry>> {
@@ -369,6 +275,23 @@ impl Outbox {
                ON articles(source_id, status);
            CREATE INDEX IF NOT EXISTS articles_claimed_at_created_at_idx
                ON articles(claimed_at, created_at);
+
+            CREATE TABLE IF NOT EXISTS delivery_intents (
+                run_id TEXT NOT NULL,
+                article_hash TEXT NOT NULL,
+                agent_id TEXT NOT NULL,
+                source_id TEXT NOT NULL,
+                started_at TEXT NOT NULL,
+                status TEXT NOT NULL DEFAULT 'pending'
+                    CHECK (status IN ('pending', 'completed', 'failed')),
+                created_at TEXT NOT NULL,
+                queued_at TEXT,
+                completed_at TEXT,
+                PRIMARY KEY (run_id, article_hash),
+                FOREIGN KEY (article_hash) REFERENCES articles(hash) ON DELETE CASCADE
+            );
+            CREATE INDEX IF NOT EXISTS delivery_intents_pending_idx
+                ON delivery_intents(status, queued_at, created_at);
            "#,
        )?;
        Ok(())
@@ -0,0 +1,110 @@
+use chrono::Utc;
+use rusqlite::{TransactionBehavior, params};
+
+use crate::{
+    domain::{Article, ArticleHash, SourceId},
+    error::Result,
+};
+
+use super::{
+    DeliveryIntent, DeliveryStatus, Outbox, parse_sql_date, persistence::save_article,
+    sql_conversion,
+};
+
+impl Outbox {
+    pub fn save_with_delivery_intent(
+        &self,
+        article: &Article,
+        intent: &DeliveryIntent,
+    ) -> Result<DeliveryStatus> {
+        let mut connection = self.connection()?;
+        let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
+        let status = save_article(&transaction, article)?;
+        if status != DeliveryStatus::Forwarded {
+            transaction.execute(
+                r#"INSERT INTO delivery_intents (
+                       run_id, article_hash, agent_id, source_id, started_at,
+                       status, created_at, queued_at, completed_at
+                   ) VALUES (?1, ?2, ?3, ?4, ?5, 'pending', ?6, NULL, NULL)
+                   ON CONFLICT(run_id, article_hash) DO NOTHING"#,
+                params![
+                    intent.run_id.as_str(),
+                    intent.article_hash.as_str(),
+                    intent.agent_id.as_str(),
+                    intent.source_id.as_str(),
+                    intent.started_at.to_rfc3339(),
+                    Utc::now().to_rfc3339(),
+                ],
+            )?;
+        }
+        transaction.commit()?;
+        Ok(status)
+    }
+
+    pub fn pending_delivery_intents(&self, limit: usize) -> Result<Vec<DeliveryIntent>> {
+        let connection = self.connection()?;
+        let mut statement = connection.prepare(
+            r#"SELECT run_id, agent_id, source_id, article_hash, started_at
+               FROM delivery_intents
+               WHERE status = 'pending'
+               ORDER BY created_at ASC LIMIT ?1"#,
+        )?;
+        statement
+            .query_map([limit as i64], |row| {
+                let source_id = SourceId::new(row.get::<_, String>(2)?)
+                    .map_err(|error| sql_conversion(2, error))?;
+                let article_hash = ArticleHash::new(row.get::<_, String>(3)?)
+                    .map_err(|error| sql_conversion(3, error))?;
+                Ok(DeliveryIntent {
+                    run_id: row.get(0)?,
+                    agent_id: row.get(1)?,
+                    source_id,
+                    article_hash,
+                    started_at: parse_sql_date(row, 4)?,
+                })
+            })?
+            .collect::<std::result::Result<Vec<_>, _>>()
+            .map_err(Into::into)
+    }
+
+    pub fn mark_delivery_intent_queued(&self, run_id: &str, hash: &str) -> Result<()> {
+        self.connection()?.execute(
+            r#"UPDATE delivery_intents SET queued_at = ?1
+               WHERE run_id = ?2 AND article_hash = ?3 AND status = 'pending'"#,
+            params![Utc::now().to_rfc3339(), run_id, hash],
+        )?;
+        Ok(())
+    }
+
+    pub fn has_delivery_intent(&self, run_id: &str, hash: &str) -> Result<bool> {
+        self.connection()?
+            .query_row(
+                r#"SELECT EXISTS(
+                       SELECT 1 FROM delivery_intents
+                       WHERE run_id = ?1 AND article_hash = ?2
+                   )"#,
+                params![run_id, hash],
+                |row| row.get(0),
+            )
+            .map_err(Into::into)
+    }
+
+    pub fn complete_delivery_intent(
+        &self,
+        run_id: &str,
+        hash: &str,
+        succeeded: bool,
+    ) -> Result<()> {
+        self.connection()?.execute(
+            r#"UPDATE delivery_intents SET status = ?1, completed_at = ?2
+               WHERE run_id = ?3 AND article_hash = ?4"#,
+            params![
+                if succeeded { "completed" } else { "failed" },
+                Utc::now().to_rfc3339(),
+                run_id,
+                hash
+            ],
+        )?;
+        Ok(())
+    }
+}
@@ -0,0 +1,65 @@
+use std::str::FromStr;
+
+use chrono::{DateTime, Utc};
+
+use crate::{
+    domain::{Article, ArticleHash, SourceId},
+    error::{CrawlError, Result},
+};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum DeliveryStatus {
+    Pending,
+    Forwarded,
+    Failed,
+}
+
+impl FromStr for DeliveryStatus {
+    type Err = CrawlError;
+
+    fn from_str(value: &str) -> Result<Self> {
+        match value {
+            "pending" => Ok(Self::Pending),
+            "forwarded" => Ok(Self::Forwarded),
+            "failed" => Ok(Self::Failed),
+            other => Err(CrawlError::Configuration(format!(
+                "unknown outbox status '{other}'"
+            ))),
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct OutboxEntry {
+    pub article: Article,
+    pub status: DeliveryStatus,
+    pub attempts: u32,
+    pub retryable: bool,
+    pub last_error: Option<String>,
+    pub created_at: DateTime<Utc>,
+    pub updated_at: DateTime<Utc>,
+    pub forwarded_at: Option<DateTime<Utc>>,
+    pub claimed_at: Option<DateTime<Utc>>,
+    pub claimed_by: Option<String>,
+}
+
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct OutboxStats {
+    pub total: usize,
+    pub pending: usize,
+    pub forwarded: usize,
+    pub failed: usize,
+    pub retryable_failed: usize,
+    pub claimed: usize,
+    pub delivery_intents_pending: usize,
+    pub delivery_intents_failed: usize,
+}
+
+#[derive(Debug, Clone)]
+pub struct DeliveryIntent {
+    pub run_id: String,
+    pub agent_id: String,
+    pub source_id: SourceId,
+    pub article_hash: ArticleHash,
+    pub started_at: DateTime<Utc>,
+}
@@ -0,0 +1,66 @@
+use chrono::Utc;
+use rusqlite::{Connection, named_params};
+
+use crate::{domain::Article, error::Result};
+
+use super::DeliveryStatus;
+
+pub(super) fn save_article(connection: &Connection, article: &Article) -> Result<DeliveryStatus> {
+    let timestamp = Utc::now().to_rfc3339();
+    let categories = serde_json::to_string(&article.categories)?;
+    let metadata = article
+        .metadata
+        .as_ref()
+        .map(serde_json::to_string)
+        .transpose()?;
+    let payload = serde_json::to_string(article)?;
+
+    connection.execute(
+        r#"
+        INSERT INTO articles (
+            hash, source_id, link, title, body, categories, metadata,
+            published_at, payload, status, attempts, retryable,
+            last_error, created_at, updated_at, forwarded_at
+        )
+        VALUES (
+            :hash, :source_id, :link, :title, :body, :categories, :metadata,
+            :published_at, :payload, 'pending', 0, 1, NULL, :now, :now, NULL
+        )
+        ON CONFLICT(hash) DO UPDATE SET
+            source_id = excluded.source_id,
+            link = excluded.link,
+            title = excluded.title,
+            body = excluded.body,
+            categories = excluded.categories,
+            metadata = excluded.metadata,
+            published_at = excluded.published_at,
+            payload = excluded.payload,
+            status = CASE WHEN articles.status = 'forwarded' THEN 'forwarded' ELSE 'pending' END,
+            last_error = CASE WHEN articles.status = 'forwarded' THEN articles.last_error ELSE NULL END,
+            retryable = CASE WHEN articles.status = 'forwarded' THEN articles.retryable ELSE 1 END,
+            updated_at = excluded.updated_at,
+            forwarded_at = CASE WHEN articles.status = 'forwarded' THEN articles.forwarded_at ELSE NULL END,
+            claimed_at = NULL,
+            claimed_by = NULL
+        "#,
+        named_params! {
+            ":hash": article.hash.as_str(),
+            ":source_id": article.source_id.as_str(),
+            ":link": article.link.as_str(),
+            ":title": article.title,
+            ":body": article.body,
+            ":categories": categories,
+            ":metadata": metadata,
+            ":published_at": article.published_at.to_rfc3339(),
+            ":payload": payload,
+            ":now": timestamp,
+        },
+    )?;
+
+    let status: String = connection.query_row(
+        "SELECT status FROM articles WHERE hash = ?1",
+        [article.hash.as_str()],
+        |row| row.get(0),
+    )?;
+    status.parse()
+}
@@ -38,7 +38,7 @@ enum Command {
    Crawl(CrawlArgs),
    /// Place one or more source discovery jobs in BullMQ.
    Schedule(ScheduleArgs),
-    /// Process BullMQ discovery and article jobs until interrupted.
+    /// Process BullMQ discovery, article, and delivery jobs until interrupted.
    Worker(WorkerArgs),
    /// Deliver pending or failed articles from the SQLite outbox.
    #[command(alias = "push")]
@@ -92,7 +92,7 @@ struct ScheduleArgs {

 #[derive(Debug, Args)]
 struct WorkerArgs {
-    /// Queue suffix to process; repeat to select both explicitly.
+    /// Queue suffix to process; repeat to select stages explicitly.
    #[arg(long, short = 'q')]
    queue: Vec<String>,
    /// Maximum number of jobs processed concurrently.
@@ -179,6 +179,10 @@ fn print_status(status: &CrawlerStatus) {
                outbox.total, outbox.pending, outbox.forwarded, outbox.failed, outbox.claimed
            );
            println!("  Retryable failures: {}", outbox.retryable_failed);
+            println!(
+                "  Delivery intents: {} pending | {} failed",
+                outbox.delivery_intents_pending, outbox.delivery_intents_failed
+            );
        }
        Err(error) => println!("  State:  unavailable ({error})"),
    }
@@ -219,6 +223,10 @@ fn print_status(status: &CrawlerStatus) {
                    "      discovered {} | processed {} | persisted {} | delivered {} | failed {}",
                    run.discovered, run.processed, run.persisted, run.delivered, run.failed
                );
+                println!(
+                    "      delivery jobs {} expected | {} processed",
+                    run.deliveries_expected, run.deliveries_processed
+                );
            }
        }
        Err(error) => println!("  State:  unavailable ({error})"),
@@ -25,6 +25,10 @@ pub(super) fn apply(config: &mut CrawlerConfig) -> Result<()> {
        "BASANGO_CRAWLER_QUEUE_ARTICLES",
        &mut config.queue.queues.articles,
    );
+    set_string(
+        "BASANGO_CRAWLER_QUEUE_DELIVERY",
+        &mut config.queue.queues.delivery,
+    );
    set_parsed(
        "BASANGO_CRAWLER_RETAIN_COMPLETED",
        &mut config.queue.retention.completed,
@@ -25,6 +25,7 @@ impl Default for QueueConfig {
 pub struct QueueNames {
    pub discovery: String,
    pub articles: String,
+    pub delivery: String,
 }

 impl Default for QueueNames {
@@ -32,6 +33,7 @@ impl Default for QueueNames {
        Self {
            discovery: "discovery".into(),
            articles: "articles".into(),
+            delivery: "delivery".into(),
        }
    }
 }
@@ -73,6 +73,7 @@ fn queue_names_schema() -> ObjectSchema {
    object()
        .optional_field("discovery", non_blank_string())
        .optional_field("articles", non_blank_string())
+        .optional_field("delivery", non_blank_string())
        .strict()
 }

@@ -7,9 +7,17 @@ use super::{CrawlerConfig, SourceConfig, schema};
 pub(super) fn validate(config: &CrawlerConfig) -> Result<()> {
    schema::validate(&serde_json::to_value(config)?)?;

-    if config.queue.queues.discovery == config.queue.queues.articles {
+    let queue_names = [
+        config.queue.queues.discovery.as_str(),
+        config.queue.queues.articles.as_str(),
+        config.queue.queues.delivery.as_str(),
+    ];
+    if queue_names[0] == queue_names[1]
+        || queue_names[0] == queue_names[2]
+        || queue_names[1] == queue_names[2]
+    {
        return Err(CrawlError::Configuration(
-            "discovery and article queue names must be distinct".into(),
+            "discovery, article, and delivery queue names must be distinct".into(),
        ));
    }
    if config.ingestion.endpoint.is_some() && config.ingestion.token.trim().is_empty() {
@@ -21,6 +21,7 @@ pub struct AgentResetReport {
    pub agent_id: String,
    pub discovery_queue: String,
    pub articles_queue: String,
+    pub delivery_queue: String,
    pub progress_trackers_removed: usize,
    pub outbox_articles_removed: usize,
 }
@@ -46,6 +47,8 @@ pub struct OpenRunStatus {
    pub started_at: DateTime<Utc>,
    pub discovered: usize,
    pub processed: usize,
+    pub deliveries_expected: usize,
+    pub deliveries_processed: usize,
    pub persisted: usize,
    pub delivered: usize,
    pub failed: usize,
@@ -180,7 +183,9 @@ impl Crawler {
                        source_id: run.source_id.to_string(),
                        started_at: run.run.started_at,
                        discovered: run.metrics.articles_discovered,
-                        processed: run.processed,
+                        processed: run.articles_processed,
+                        deliveries_expected: run.deliveries_expected,
+                        deliveries_processed: run.deliveries_processed,
                        persisted: run.metrics.articles_persisted,
                        delivered: run.metrics.articles_delivered,
                        failed: run.metrics.articles_failed,
@@ -206,6 +211,7 @@ impl Crawler {
            agent_id,
            discovery_queue,
            articles_queue,
+            delivery_queue,
            progress_trackers_removed,
        } = queue.reset_agent().await?;
        let outbox = Outbox::open(&self.runtime.config.sqlite_path(), true)?;
@@ -221,6 +227,7 @@ impl Crawler {
            agent_id,
            discovery_queue,
            articles_queue,
+            delivery_queue,
            progress_trackers_removed,
            outbox_articles_removed,
        })
@@ -1,277 +0,0 @@
-//! Domain types: the vocabulary of the crawler.
-//!
-//! Domain values describe *what* Basango works with. They deliberately do not
-//! know how HTTP, Redis, SQLite, or the CLI work. This dependency direction is
-//! what lets the same types move through synchronous and queued execution.
-
-use std::{fmt, str::FromStr};
-
-use chrono::{DateTime, NaiveDate, TimeZone, Utc};
-use serde::{Deserialize, Serialize};
-use url::Url;
-
-use crate::error::{CrawlError, Result};
-
-// --- Source identity -------------------------------------------------------
-
-/// A validated source identifier.
-///
-/// Using a newtype prevents an arbitrary or empty `String` from being passed
-/// wherever the crawler expects the identity of a configured source.
-#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)]
-#[serde(transparent)]
-pub struct SourceId(String);
-
-impl SourceId {
-    pub fn new(value: impl Into<String>) -> Result<Self> {
-        let value = value.into();
-        let normalized = value.trim();
-        if normalized.is_empty() {
-            return Err(CrawlError::Configuration(
-                "source id cannot be empty".into(),
-            ));
-        }
-        Ok(Self(normalized.to_owned()))
-    }
-
-    pub fn as_str(&self) -> &str {
-        &self.0
-    }
-}
-
-impl fmt::Display for SourceId {
-    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
-        formatter.write_str(&self.0)
-    }
-}
-
-impl FromStr for SourceId {
-    type Err = CrawlError;
-
-    fn from_str(value: &str) -> Result<Self> {
-        Self::new(value)
-    }
-}
-
-impl AsRef<str> for SourceId {
-    fn as_ref(&self) -> &str {
-        self.as_str()
-    }
-}
-
-impl<'de> Deserialize<'de> for SourceId {
-    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
-    where
-        D: serde::Deserializer<'de>,
-    {
-        let value = String::deserialize(deserializer)?;
-        Self::new(value).map_err(serde::de::Error::custom)
-    }
-}
-
-// --- Articles -------------------------------------------------------------
-
-/// Optional metadata discovered from Open Graph or WordPress fields.
-#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(rename_all = "camelCase")]
-pub struct ArticleMetadata {
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub url: Option<Url>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub title: Option<String>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub author: Option<String>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub description: Option<String>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub image: Option<Url>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub published_at: Option<String>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub updated_at: Option<String>,
-}
-
-impl ArticleMetadata {
-    /// Empty metadata is represented as `None` instead of an object full of
-    /// `null` values. That makes absence explicit to downstream code.
-    pub fn is_empty(&self) -> bool {
-        self.url.is_none()
-            && self.title.is_none()
-            && self.author.is_none()
-            && self.description.is_none()
-            && self.image.is_none()
-            && self.published_at.is_none()
-            && self.updated_at.is_none()
-    }
-}
-
-/// A source crawler's output before normalization and hashing.
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct ArticleDraft {
-    pub title: String,
-    pub body: String,
-    pub link: Url,
-    pub source_id: SourceId,
-    pub categories: Vec<String>,
-    pub metadata: Option<ArticleMetadata>,
-    pub published_at: DateTime<Utc>,
-}
-
-/// The validated representation persisted in the outbox and sent to the API.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(rename_all = "camelCase")]
-pub struct Article {
-    pub hash: String,
-    pub title: String,
-    pub body: String,
-    pub link: Url,
-    pub source_id: SourceId,
-    #[serde(default)]
-    pub categories: Vec<String>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub metadata: Option<ArticleMetadata>,
-    pub published_at: DateTime<Utc>,
-}
-
-// --- Ranges and crawl options --------------------------------------------
-
-/// Inclusive page boundaries. HTML sources may start at page zero, whereas
-/// WordPress starts at page one, so zero is a valid value here.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
-pub struct PageRange {
-    pub start: u32,
-    pub end: u32,
-}
-
-impl PageRange {
-    pub fn new(start: u32, end: u32) -> Result<Self> {
-        if end < start {
-            return Err(CrawlError::InvalidRange(format!(
-                "end page {end} is before start page {start}"
-            )));
-        }
-        Ok(Self { start, end })
-    }
-
-    /// Parse the CLI representation `start:end`.
-    pub fn parse(spec: &str) -> Result<Self> {
-        let (start, end) = split_range(spec, "page")?;
-        Self::new(
-            start
-                .parse()
-                .map_err(|_| CrawlError::InvalidRange(format!("invalid start page '{start}'")))?,
-            end.parse()
-                .map_err(|_| CrawlError::InvalidRange(format!("invalid end page '{end}'")))?,
-        )
-    }
-}
-
-impl fmt::Display for PageRange {
-    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(formatter, "{}:{}", self.start, self.end)
-    }
-}
-
-/// Inclusive UTC time boundaries used to filter articles.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
-pub struct DateRange {
-    pub start: DateTime<Utc>,
-    pub end: DateTime<Utc>,
-}
-
-impl DateRange {
-    pub fn new(start: DateTime<Utc>, end: DateTime<Utc>) -> Result<Self> {
-        if end < start {
-            return Err(CrawlError::InvalidRange(
-                "end date must be on or after start date".into(),
-            ));
-        }
-        Ok(Self { start, end })
-    }
-
-    /// Parse `YYYY-MM-DD:YYYY-MM-DD`. The end date is expanded to the last
-    /// nanosecond of that day so the range behaves as users expect.
-    pub fn parse(spec: &str) -> Result<Self> {
-        let (start, end) = split_range(spec, "date")?;
-        let start = parse_date(start)?
-            .and_hms_opt(0, 0, 0)
-            .expect("midnight is valid");
-        let end = parse_date(end)?
-            .and_hms_nano_opt(23, 59, 59, 999_999_999)
-            .expect("end of day is valid");
-        Self::new(Utc.from_utc_datetime(&start), Utc.from_utc_datetime(&end))
-    }
-
-    pub fn contains(&self, timestamp: DateTime<Utc>) -> bool {
-        self.start <= timestamp && timestamp <= self.end
-    }
-
-    /// Crawlers receive newest-first listings. Once an article is older than
-    /// `start`, all following items are normally older too and crawling can stop.
-    pub fn is_older_than_range(&self, timestamp: DateTime<Utc>) -> bool {
-        timestamp < self.start
-    }
-}
-
-impl fmt::Display for DateRange {
-    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(
-            formatter,
-            "{}:{}",
-            self.start.format("%Y-%m-%d"),
-            self.end.format("%Y-%m-%d")
-        )
-    }
-}
-
-/// One request to crawl a configured source.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(rename_all = "camelCase")]
-pub struct CrawlRequest {
-    pub source_id: SourceId,
-    pub page_range: Option<PageRange>,
-    pub date_range: Option<DateRange>,
-    pub category: Option<String>,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub direction: Option<UpdateDirection>,
-}
-
-/// Crawling direction used when the backend supplies an update boundary.
-#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(rename_all = "lowercase")]
-pub enum UpdateDirection {
-    Backward,
-    #[default]
-    Forward,
-}
-
-impl FromStr for UpdateDirection {
-    type Err = CrawlError;
-
-    fn from_str(value: &str) -> Result<Self> {
-        match value.trim().to_ascii_lowercase().as_str() {
-            "backward" => Ok(Self::Backward),
-            "forward" => Ok(Self::Forward),
-            _ => Err(CrawlError::InvalidRange(format!(
-                "invalid update direction '{value}'; expected forward or backward"
-            ))),
-        }
-    }
-}
-
-// --- Helpers --------------------------------------------------------------
-
-fn split_range<'a>(spec: &'a str, kind: &str) -> Result<(&'a str, &'a str)> {
-    spec.split_once(':').ok_or_else(|| {
-        CrawlError::InvalidRange(format!("invalid {kind} range '{spec}'; expected start:end"))
-    })
-}
-
-fn parse_date(value: &str) -> Result<NaiveDate> {
-    NaiveDate::parse_from_str(value, "%Y-%m-%d")
-        .map_err(|_| CrawlError::InvalidRange(format!("invalid date '{value}'; use YYYY-MM-DD")))
-}
-
-#[cfg(test)]
-#[path = "../tests/unit/domain.rs"]
-mod tests;
@@ -0,0 +1,123 @@
+use std::{fmt, ops::Deref};
+
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+use url::Url;
+
+use super::SourceId;
+use crate::error::{CrawlError, Result};
+
+/// The stable identity of an article URL in the local outbox and ingestion API.
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)]
+#[serde(transparent)]
+pub struct ArticleHash(String);
+
+impl ArticleHash {
+    pub fn from_url(url: &Url) -> Self {
+        Self(format!("{:x}", md5::compute(url.as_str())))
+    }
+
+    pub fn new(value: impl Into<String>) -> Result<Self> {
+        let value = value.into();
+        if value.len() != 32 || !value.bytes().all(|byte| byte.is_ascii_hexdigit()) {
+            return Err(CrawlError::InvalidArticle(format!(
+                "invalid article hash '{value}'"
+            )));
+        }
+        Ok(Self(value.to_ascii_lowercase()))
+    }
+
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+impl fmt::Display for ArticleHash {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str(&self.0)
+    }
+}
+
+impl AsRef<str> for ArticleHash {
+    fn as_ref(&self) -> &str {
+        self.as_str()
+    }
+}
+
+impl Deref for ArticleHash {
+    type Target = str;
+
+    fn deref(&self) -> &Self::Target {
+        self.as_str()
+    }
+}
+
+impl<'de> Deserialize<'de> for ArticleHash {
+    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        let value = String::deserialize(deserializer)?;
+        Self::new(value).map_err(serde::de::Error::custom)
+    }
+}
+
+/// Optional metadata discovered from Open Graph or WordPress fields.
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ArticleMetadata {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub url: Option<Url>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub title: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub author: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub description: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub image: Option<Url>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub published_at: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub updated_at: Option<String>,
+}
+
+impl ArticleMetadata {
+    pub fn is_empty(&self) -> bool {
+        self.url.is_none()
+            && self.title.is_none()
+            && self.author.is_none()
+            && self.description.is_none()
+            && self.image.is_none()
+            && self.published_at.is_none()
+            && self.updated_at.is_none()
+    }
+}
+
+/// A source crawler's output before normalization and hashing.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ArticleDraft {
+    pub title: String,
+    pub body: String,
+    pub link: Url,
+    pub source_id: SourceId,
+    pub categories: Vec<String>,
+    pub metadata: Option<ArticleMetadata>,
+    pub published_at: DateTime<Utc>,
+}
+
+/// The validated representation persisted locally and sent to the API.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Article {
+    pub hash: ArticleHash,
+    pub title: String,
+    pub body: String,
+    pub link: Url,
+    pub source_id: SourceId,
+    #[serde(default)]
+    pub categories: Vec<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub metadata: Option<ArticleMetadata>,
+    pub published_at: DateTime<Utc>,
+}
@@ -0,0 +1,132 @@
+use std::{fmt, str::FromStr};
+
+use chrono::{DateTime, NaiveDate, TimeZone, Utc};
+use serde::{Deserialize, Serialize};
+
+use super::SourceId;
+use crate::error::{CrawlError, Result};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub struct PageRange {
+    pub start: u32,
+    pub end: u32,
+}
+
+impl PageRange {
+    pub fn new(start: u32, end: u32) -> Result<Self> {
+        if end < start {
+            return Err(CrawlError::InvalidRange(format!(
+                "end page {end} is before start page {start}"
+            )));
+        }
+        Ok(Self { start, end })
+    }
+
+    pub fn parse(spec: &str) -> Result<Self> {
+        let (start, end) = split_range(spec, "page")?;
+        Self::new(
+            start
+                .parse()
+                .map_err(|_| CrawlError::InvalidRange(format!("invalid start page '{start}'")))?,
+            end.parse()
+                .map_err(|_| CrawlError::InvalidRange(format!("invalid end page '{end}'")))?,
+        )
+    }
+}
+
+impl fmt::Display for PageRange {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(formatter, "{}:{}", self.start, self.end)
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub struct DateRange {
+    pub start: DateTime<Utc>,
+    pub end: DateTime<Utc>,
+}
+
+impl DateRange {
+    pub fn new(start: DateTime<Utc>, end: DateTime<Utc>) -> Result<Self> {
+        if end < start {
+            return Err(CrawlError::InvalidRange(
+                "end date must be on or after start date".into(),
+            ));
+        }
+        Ok(Self { start, end })
+    }
+
+    pub fn parse(spec: &str) -> Result<Self> {
+        let (start, end) = split_range(spec, "date")?;
+        let start = parse_date(start)?
+            .and_hms_opt(0, 0, 0)
+            .expect("midnight is valid");
+        let end = parse_date(end)?
+            .and_hms_nano_opt(23, 59, 59, 999_999_999)
+            .expect("end of day is valid");
+        Self::new(Utc.from_utc_datetime(&start), Utc.from_utc_datetime(&end))
+    }
+
+    pub fn contains(&self, timestamp: DateTime<Utc>) -> bool {
+        self.start <= timestamp && timestamp <= self.end
+    }
+
+    pub fn is_older_than_range(&self, timestamp: DateTime<Utc>) -> bool {
+        timestamp < self.start
+    }
+}
+
+impl fmt::Display for DateRange {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            formatter,
+            "{}:{}",
+            self.start.format("%Y-%m-%d"),
+            self.end.format("%Y-%m-%d")
+        )
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CrawlRequest {
+    pub source_id: SourceId,
+    pub page_range: Option<PageRange>,
+    pub date_range: Option<DateRange>,
+    pub category: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub direction: Option<UpdateDirection>,
+}
+
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum UpdateDirection {
+    Backward,
+    #[default]
+    Forward,
+}
+
+impl FromStr for UpdateDirection {
+    type Err = CrawlError;
+
+    fn from_str(value: &str) -> Result<Self> {
+        match value.trim().to_ascii_lowercase().as_str() {
+            "backward" => Ok(Self::Backward),
+            "forward" => Ok(Self::Forward),
+            _ => Err(CrawlError::InvalidRange(format!(
+                "invalid update direction '{value}'; expected forward or backward"
+            ))),
+        }
+    }
+}
+
+fn split_range<'a>(spec: &'a str, kind: &str) -> Result<(&'a str, &'a str)> {
+    spec.split_once(':').ok_or_else(|| {
+        CrawlError::InvalidRange(format!("invalid {kind} range '{spec}'; expected start:end"))
+    })
+}
+
+fn parse_date(value: &str) -> Result<NaiveDate> {
+    NaiveDate::parse_from_str(value, "%Y-%m-%d")
+        .map_err(|_| CrawlError::InvalidRange(format!("invalid date '{value}'; use YYYY-MM-DD")))
+}
@@ -0,0 +1,25 @@
+/// Durable state of an article's delivery to the ingestion API.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum DeliveryState {
+    Pending,
+    Forwarded,
+    Failed,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RetryDecision {
+    Retry,
+    Stop,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum DeliveryOutcome {
+    Delivered {
+        status: u16,
+    },
+    Failed {
+        decision: RetryDecision,
+        status: Option<u16>,
+        message: String,
+    },
+}
@@ -0,0 +1,20 @@
+//! Domain types: the vocabulary of the crawler.
+//!
+//! These values describe what Basango works with and deliberately avoid HTTP,
+//! Redis, SQLite, and CLI concerns.
+
+mod article;
+mod crawl;
+mod delivery;
+mod run;
+mod source;
+
+pub use article::{Article, ArticleDraft, ArticleHash, ArticleMetadata};
+pub use crawl::{CrawlRequest, DateRange, PageRange, UpdateDirection};
+pub use delivery::{DeliveryOutcome, DeliveryState, RetryDecision};
+pub use run::{AgentId, RunId};
+pub use source::{CategorySlug, SourceId};
+
+#[cfg(test)]
+#[path = "../../tests/unit/domain.rs"]
+mod tests;
@@ -0,0 +1,82 @@
+use std::fmt;
+
+use serde::{Deserialize, Serialize};
+use uuid::Uuid;
+
+use crate::error::{CrawlError, Result};
+
+macro_rules! string_id {
+    ($name:ident, $label:literal) => {
+        #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)]
+        #[serde(transparent)]
+        pub struct $name(String);
+
+        impl $name {
+            pub fn new(value: impl Into<String>) -> Result<Self> {
+                let value = value.into();
+                let value = value.trim();
+                if value.is_empty() {
+                    return Err(CrawlError::Configuration(format!(
+                        "{} cannot be empty",
+                        $label
+                    )));
+                }
+                Ok(Self(value.to_owned()))
+            }
+
+            pub fn as_str(&self) -> &str {
+                &self.0
+            }
+        }
+
+        impl fmt::Display for $name {
+            fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+                formatter.write_str(&self.0)
+            }
+        }
+
+        impl<'de> Deserialize<'de> for $name {
+            fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+            where
+                D: serde::Deserializer<'de>,
+            {
+                let value = String::deserialize(deserializer)?;
+                Self::new(value).map_err(serde::de::Error::custom)
+            }
+        }
+    };
+}
+
+string_id!(AgentId, "agent id");
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[serde(transparent)]
+pub struct RunId(Uuid);
+
+impl RunId {
+    pub fn new() -> Self {
+        Self(Uuid::now_v7())
+    }
+
+    pub fn parse(value: &str) -> Result<Self> {
+        Uuid::parse_str(value)
+            .map(Self)
+            .map_err(|_| CrawlError::Configuration(format!("invalid run id '{value}'")))
+    }
+
+    pub fn as_uuid(&self) -> Uuid {
+        self.0
+    }
+}
+
+impl Default for RunId {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl fmt::Display for RunId {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.0.fmt(formatter)
+    }
+}
@@ -0,0 +1,103 @@
+use std::{fmt, str::FromStr};
+
+use serde::{Deserialize, Serialize};
+
+use crate::error::{CrawlError, Result};
+
+/// A validated source identifier.
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)]
+#[serde(transparent)]
+pub struct SourceId(String);
+
+impl SourceId {
+    pub fn new(value: impl Into<String>) -> Result<Self> {
+        let value = value.into();
+        let normalized = value.trim();
+        if normalized.is_empty() {
+            return Err(CrawlError::Configuration(
+                "source id cannot be empty".into(),
+            ));
+        }
+        Ok(Self(normalized.to_owned()))
+    }
+
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+impl fmt::Display for SourceId {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str(&self.0)
+    }
+}
+
+impl FromStr for SourceId {
+    type Err = CrawlError;
+
+    fn from_str(value: &str) -> Result<Self> {
+        Self::new(value)
+    }
+}
+
+impl AsRef<str> for SourceId {
+    fn as_ref(&self) -> &str {
+        self.as_str()
+    }
+}
+
+impl<'de> Deserialize<'de> for SourceId {
+    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        let value = String::deserialize(deserializer)?;
+        Self::new(value).map_err(serde::de::Error::custom)
+    }
+}
+
+/// A normalized category path used by category-indexed sources.
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)]
+#[serde(transparent)]
+pub struct CategorySlug(String);
+
+impl CategorySlug {
+    pub fn new(value: impl Into<String>) -> Result<Self> {
+        let value = value.into();
+        let normalized = value.trim().to_lowercase();
+        if normalized.is_empty() {
+            return Err(CrawlError::Configuration(
+                "category slug cannot be empty".into(),
+            ));
+        }
+        Ok(Self(normalized))
+    }
+
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+impl fmt::Display for CategorySlug {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str(&self.0)
+    }
+}
+
+impl FromStr for CategorySlug {
+    type Err = CrawlError;
+
+    fn from_str(value: &str) -> Result<Self> {
+        Self::new(value)
+    }
+}
+
+impl<'de> Deserialize<'de> for CategorySlug {
+    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        let value = String::deserialize(deserializer)?;
+        Self::new(value).map_err(serde::de::Error::custom)
+    }
+}
@@ -8,8 +8,8 @@ mod sync;
 mod worker;

 pub(crate) use queue::{
-    AgentResetReport as QueueResetReport, DiscoverJob, FetchJob, JobQueue, QueueSnapshot,
-    QueuedRunContext,
+    AgentResetReport as QueueResetReport, DeliveryJob, DiscoverJob, FetchJob, JobQueue,
+    QueueSnapshot, QueuedRunContext,
 };
 pub use sync::CrawlReport;
 pub(crate) use sync::{crawl_now, forward_pending};
@@ -0,0 +1,181 @@
+//! BullMQ-backed jobs used by `schedule` and `worker`.
+//!
+//! BullMQ owns queue state transitions, retry backoff, job locks, lock renewal,
+//! stalled-job recovery, and retention. This module defines Basango's typed
+//! payloads and translates crawler configuration into BullMQ options.
+
+mod model;
+mod progress;
+mod support;
+
+use bullmq::Queue;
+use bullmq::types::BackoffStrategy;
+
+use crate::{
+    config::QueueConfig,
+    error::{CrawlError, Result},
+};
+
+pub use model::{
+    AgentResetReport, DeliveryJob, DiscoverJob, FetchJob, OpenQueuedRun, QueueSnapshot,
+    QueuedRunContext, QueuedRunUpdate,
+};
+pub(crate) use support::redis_options;
+use support::{encode_agent_id, queue_options, retention, scoped_queue_name, stable_job_id};
+
+const JOB_ATTEMPTS: u32 = 3;
+const RETRY_DELAY_MS: u64 = 1_000;
+const RUN_PROGRESS_TTL_SECONDS: usize = 7 * 24 * 60 * 60;
+
+/// Producer-side access to the crawler queues.
+pub struct JobQueue {
+    pub(super) discovery: Queue,
+    pub(super) articles: Queue,
+    pub(super) delivery: Queue,
+    pub(super) config: QueueConfig,
+    pub(super) agent_id: String,
+    pub(super) agent_scope: String,
+    pub(super) discovery_name: String,
+    pub(super) articles_name: String,
+    pub(super) delivery_name: String,
+    pub(super) progress_client: redis::Client,
+}
+
+impl JobQueue {
+    pub async fn connect(config: &QueueConfig, agent_id: &str) -> Result<Self> {
+        let options = queue_options(config);
+        let progress_client = redis::Client::open(config.redis_url.clone())?;
+        let agent_scope = encode_agent_id(agent_id);
+        let discovery_name = scoped_queue_name(&agent_scope, &config.queues.discovery);
+        let articles_name = scoped_queue_name(&agent_scope, &config.queues.articles);
+        let delivery_name = scoped_queue_name(&agent_scope, &config.queues.delivery);
+        let (discovery, articles, delivery) = tokio::try_join!(
+            Queue::with_options(&discovery_name, options.clone()),
+            Queue::with_options(&articles_name, options.clone()),
+            Queue::with_options(&delivery_name, options),
+        )?;
+        Ok(Self {
+            discovery,
+            articles,
+            delivery,
+            config: config.clone(),
+            agent_id: agent_id.to_owned(),
+            agent_scope,
+            discovery_name,
+            articles_name,
+            delivery_name,
+            progress_client,
+        })
+    }
+
+    pub fn names(&self) -> [&str; 3] {
+        [
+            self.discovery_name.as_str(),
+            self.articles_name.as_str(),
+            self.delivery_name.as_str(),
+        ]
+    }
+
+    pub fn validate_names(&self, names: &[String]) -> Result<()> {
+        let scoped = self.names();
+        let base = [
+            self.config.queues.discovery.as_str(),
+            self.config.queues.articles.as_str(),
+            self.config.queues.delivery.as_str(),
+        ];
+        for name in names {
+            if !scoped.contains(&name.as_str()) && !base.contains(&name.as_str()) {
+                return Err(CrawlError::Queue(format!(
+                    "unknown queue '{name}'; expected {}, {}, or {}",
+                    base[0], base[1], base[2]
+                )));
+            }
+        }
+        Ok(())
+    }
+
+    pub fn resolve_names(&self, names: &[String]) -> Vec<String> {
+        names
+            .iter()
+            .map(|name| {
+                if name == &self.config.queues.discovery {
+                    self.discovery_name.clone()
+                } else if name == &self.config.queues.articles {
+                    self.articles_name.clone()
+                } else if name == &self.config.queues.delivery {
+                    self.delivery_name.clone()
+                } else {
+                    name.clone()
+                }
+            })
+            .collect()
+    }
+
+    pub async fn enqueue_discovery(&self, job: DiscoverJob) -> Result<String> {
+        let id = stable_job_id("discover", &job)?;
+        let run_id = job.run.run_id.clone();
+        let source_id = job.request.source_id.clone();
+        let queued = self
+            .discovery
+            .add("discover-source", job)
+            .job_id(&id)
+            .attempts(JOB_ATTEMPTS)
+            .backoff(BackoffStrategy::Exponential(RETRY_DELAY_MS))
+            .remove_on_complete(retention(self.config.retention.completed))
+            .remove_on_fail(retention(self.config.retention.failed))
+            .await?;
+        tracing::info!(
+            agent_id = self.agent_id,
+            queue = self.discovery_name,
+            job_id = queued.id(),
+            %run_id,
+            source = %source_id,
+            "discovery job enqueued"
+        );
+        Ok(queued.id().to_owned())
+    }
+
+    pub async fn enqueue_article(&self, job: FetchJob) -> Result<String> {
+        let identity = (&job.run.run_id, &job.request.source_id, &job.article.url);
+        let id = stable_job_id("article", &identity)?;
+        let queued = self
+            .articles
+            .add("fetch-article", job)
+            .job_id(&id)
+            .attempts(JOB_ATTEMPTS)
+            .backoff(BackoffStrategy::Exponential(RETRY_DELAY_MS))
+            .remove_on_complete(retention(self.config.retention.completed))
+            .remove_on_fail(retention(self.config.retention.failed))
+            .await?;
+        Ok(queued.id().to_owned())
+    }
+
+    pub async fn enqueue_delivery(&self, job: DeliveryJob) -> Result<String> {
+        let identity = (&job.run.run_id, &job.article_hash);
+        let id = stable_job_id("delivery", &identity)?;
+        let queued = self
+            .delivery
+            .add("deliver-article", job)
+            .job_id(&id)
+            .attempts(JOB_ATTEMPTS)
+            .backoff(BackoffStrategy::Exponential(RETRY_DELAY_MS))
+            .remove_on_complete(retention(self.config.retention.completed))
+            .remove_on_fail(retention(self.config.retention.failed))
+            .await?;
+        Ok(queued.id().to_owned())
+    }
+
+    pub async fn retry_failed_deliveries(&self) -> Result<()> {
+        self.delivery.retry_jobs("failed", 1_000, None).await?;
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+use bullmq::types::{JobCounts, RemoveOnFinish};
+#[cfg(test)]
+use support::snapshot_from_counts;
+
+#[cfg(test)]
+#[path = "../../../tests/unit/execution/queue.rs"]
+mod tests;
@@ -0,0 +1,75 @@
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+
+use crate::{
+    domain::{ArticleHash, CrawlRequest, SourceId},
+    sources::ArticleSeed,
+    telemetry::RunMetrics,
+};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct QueuedRunContext {
+    pub run_id: String,
+    pub agent_id: String,
+    pub started_at: DateTime<Utc>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DiscoverJob {
+    pub request: CrawlRequest,
+    pub run: QueuedRunContext,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct FetchJob {
+    pub request: CrawlRequest,
+    pub article: ArticleSeed,
+    pub run: QueuedRunContext,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DeliveryJob {
+    pub article_hash: ArticleHash,
+    pub source_id: SourceId,
+    pub run: QueuedRunContext,
+}
+
+#[derive(Debug, Clone, Copy)]
+pub struct QueuedRunUpdate {
+    pub metrics: RunMetrics,
+    pub terminal: bool,
+}
+
+#[derive(Debug, Clone)]
+pub struct OpenQueuedRun {
+    pub run: QueuedRunContext,
+    pub source_id: SourceId,
+    pub articles_processed: usize,
+    pub deliveries_expected: usize,
+    pub deliveries_processed: usize,
+    pub metrics: RunMetrics,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct AgentResetReport {
+    pub agent_id: String,
+    pub discovery_queue: String,
+    pub articles_queue: String,
+    pub delivery_queue: String,
+    pub progress_trackers_removed: usize,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct QueueSnapshot {
+    pub name: String,
+    pub workers: usize,
+    pub waiting: u64,
+    pub active: u64,
+    pub delayed: u64,
+    pub prioritized: u64,
+    pub completed: u64,
+    pub failed: u64,
+    pub waiting_children: u64,
+    pub paused: u64,
+}
@@ -1,196 +1,14 @@
-//! BullMQ-backed jobs used by `schedule` and `worker`.
-//!
-//! BullMQ owns queue state transitions, retry backoff, job locks, lock renewal,
-//! stalled-job recovery, and retention. This module only defines Basango's
-//! typed payloads and translates crawler configuration into BullMQ options.
-
-use std::time::Duration;
-
-use bullmq::options::RedisConnectionOptions;
-use bullmq::types::{BackoffStrategy, JobCounts, KeepJobs, RemoveOnFinish};
-use bullmq::{Queue, QueueOptions};
 use chrono::{DateTime, Utc};
-use serde::{Deserialize, Serialize};

-use crate::{
-    config::QueueConfig,
-    domain::{CrawlRequest, SourceId},
-    error::{CrawlError, Result},
-    sources::ArticleSeed,
-    telemetry::RunMetrics,
+use crate::{domain::SourceId, error::Result, telemetry::RunMetrics};
+
+use super::{
+    AgentResetReport, JobQueue, OpenQueuedRun, QueueSnapshot, QueuedRunContext, QueuedRunUpdate,
+    RUN_PROGRESS_TTL_SECONDS,
+    support::{metrics_from_values, parse_metric, queue_snapshot, queued_update_from_values},
 };

-const JOB_ATTEMPTS: u32 = 3;
-const RETRY_DELAY_MS: u64 = 1_000;
-const RUN_PROGRESS_TTL_SECONDS: usize = 7 * 24 * 60 * 60;
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-#[serde(rename_all = "camelCase")]
-pub struct QueuedRunContext {
-    pub run_id: String,
-    pub agent_id: String,
-    pub started_at: DateTime<Utc>,
-}
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct DiscoverJob {
-    pub request: CrawlRequest,
-    pub run: QueuedRunContext,
-}
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct FetchJob {
-    pub request: CrawlRequest,
-    pub article: ArticleSeed,
-    pub run: QueuedRunContext,
-}
-
-#[derive(Debug, Clone, Copy)]
-pub struct QueuedRunUpdate {
-    pub metrics: RunMetrics,
-    pub terminal: bool,
-}
-
-#[derive(Debug, Clone)]
-pub struct OpenQueuedRun {
-    pub run: QueuedRunContext,
-    pub source_id: SourceId,
-    pub processed: usize,
-    pub metrics: RunMetrics,
-}
-
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct AgentResetReport {
-    pub agent_id: String,
-    pub discovery_queue: String,
-    pub articles_queue: String,
-    pub progress_trackers_removed: usize,
-}
-
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct QueueSnapshot {
-    pub name: String,
-    pub workers: usize,
-    pub waiting: u64,
-    pub active: u64,
-    pub delayed: u64,
-    pub prioritized: u64,
-    pub completed: u64,
-    pub failed: u64,
-    pub waiting_children: u64,
-    pub paused: u64,
-}
-
-/// Producer-side access to the two crawler queues.
-pub struct JobQueue {
-    discovery: Queue,
-    articles: Queue,
-    config: QueueConfig,
-    agent_id: String,
-    agent_scope: String,
-    discovery_name: String,
-    articles_name: String,
-    progress_client: redis::Client,
-}
-
 impl JobQueue {
-    pub async fn connect(config: &QueueConfig, agent_id: &str) -> Result<Self> {
-        let options = queue_options(config);
-        let progress_client = redis::Client::open(config.redis_url.clone())?;
-        let agent_scope = encode_agent_id(agent_id);
-        let discovery_name = scoped_queue_name(&agent_scope, &config.queues.discovery);
-        let articles_name = scoped_queue_name(&agent_scope, &config.queues.articles);
-        let (discovery, articles) = tokio::try_join!(
-            Queue::with_options(&discovery_name, options.clone()),
-            Queue::with_options(&articles_name, options),
-        )?;
-        Ok(Self {
-            discovery,
-            articles,
-            config: config.clone(),
-            agent_id: agent_id.to_owned(),
-            agent_scope,
-            discovery_name,
-            articles_name,
-            progress_client,
-        })
-    }
-
-    pub fn names(&self) -> [&str; 2] {
-        [self.discovery_name.as_str(), self.articles_name.as_str()]
-    }
-
-    pub fn validate_names(&self, names: &[String]) -> Result<()> {
-        let scoped = self.names();
-        let base = [
-            self.config.queues.discovery.as_str(),
-            self.config.queues.articles.as_str(),
-        ];
-        for name in names {
-            if !scoped.contains(&name.as_str()) && !base.contains(&name.as_str()) {
-                return Err(CrawlError::Queue(format!(
-                    "unknown queue '{name}'; expected {} or {}",
-                    base[0], base[1]
-                )));
-            }
-        }
-        Ok(())
-    }
-
-    pub fn resolve_names(&self, names: &[String]) -> Vec<String> {
-        names
-            .iter()
-            .map(|name| {
-                if name == &self.config.queues.discovery {
-                    self.discovery_name.clone()
-                } else if name == &self.config.queues.articles {
-                    self.articles_name.clone()
-                } else {
-                    name.clone()
-                }
-            })
-            .collect()
-    }
-
-    pub async fn enqueue_discovery(&self, job: DiscoverJob) -> Result<String> {
-        let id = stable_job_id("discover", &job)?;
-        let run_id = job.run.run_id.clone();
-        let source_id = job.request.source_id.clone();
-        let queued = self
-            .discovery
-            .add("discover-source", job)
-            .job_id(&id)
-            .attempts(JOB_ATTEMPTS)
-            .backoff(BackoffStrategy::Exponential(RETRY_DELAY_MS))
-            .remove_on_complete(retention(self.config.retention.completed))
-            .remove_on_fail(retention(self.config.retention.failed))
-            .await?;
-        tracing::info!(
-            agent_id = self.agent_id,
-            queue = self.discovery_name,
-            job_id = queued.id(),
-            %run_id,
-            source = %source_id,
-            "discovery job enqueued"
-        );
-        Ok(queued.id().to_owned())
-    }
-
-    pub async fn enqueue_article(&self, job: FetchJob) -> Result<String> {
-        let identity = (&job.run.run_id, &job.request.source_id, &job.article.url);
-        let id = stable_job_id("article", &identity)?;
-        let queued = self
-            .articles
-            .add("fetch-article", job)
-            .job_id(&id)
-            .attempts(JOB_ATTEMPTS)
-            .backoff(BackoffStrategy::Exponential(RETRY_DELAY_MS))
-            .remove_on_complete(retention(self.config.retention.completed))
-            .remove_on_fail(retention(self.config.retention.failed))
-            .await?;
-        Ok(queued.id().to_owned())
-    }
-
    pub async fn prepare_run(&self, run: &QueuedRunContext, source_id: &SourceId) -> Result<()> {
        const SCRIPT: &str = r#"
            if redis.call('EXISTS', KEYS[1]) == 0 then
@@ -199,7 +17,9 @@ impl JobQueue {
                    'startedAt', ARGV[2],
                    'discovered', 0,
                    'discoveryComplete', 0,
-                    'processed', 0,
+                    'articleProcessed', 0,
+                    'deliveryExpected', 0,
+                    'deliveryProcessed', 0,
                    'persisted', 0,
                    'delivered', 0,
                    'failed', 0,
@@ -227,15 +47,22 @@ impl JobQueue {
        &self,
        run_id: &str,
        batch_id: &str,
-        discovered: usize,
+        article_job_ids: &[String],
    ) -> Result<Option<RunMetrics>> {
        const SCRIPT: &str = r#"
            if redis.call('EXISTS', KEYS[1]) == 0 then return {} end
            if tonumber(redis.call('HGET', KEYS[1], 'terminalSent')) == 1 then return {} end
-            if redis.call('HSETNX', KEYS[1], 'batch:' .. ARGV[1], 1) == 1 then
-                redis.call('HINCRBY', KEYS[1], 'discovered', ARGV[2])
+            redis.call('HSET', KEYS[1], 'batch:' .. ARGV[1], 1)
+            local added = 0
+            for index = 3, #ARGV do
+                if redis.call('HSETNX', KEYS[1], 'article:' .. ARGV[index], 1) == 1 then
+                    added = added + 1
+                end
            end
-            redis.call('EXPIRE', KEYS[1], ARGV[3])
+            if added > 0 then
+                redis.call('HINCRBY', KEYS[1], 'discovered', added)
+            end
+            redis.call('EXPIRE', KEYS[1], ARGV[2])
            return {
                tonumber(redis.call('HGET', KEYS[1], 'discovered')) or 0,
                tonumber(redis.call('HGET', KEYS[1], 'persisted')) or 0,
@@ -251,8 +78,8 @@ impl JobQueue {
        let values: Vec<i64> = redis::Script::new(SCRIPT)
            .key(key)
            .arg(batch_id)
-            .arg(discovered)
            .arg(RUN_PROGRESS_TTL_SECONDS)
+            .arg(article_job_ids)
            .invoke_async(&mut connection)
            .await?;
        if values.is_empty() {
@@ -268,9 +95,11 @@ impl JobQueue {
            if tonumber(redis.call('HGET', KEYS[1], 'terminalSent')) == 1 then return {} end
            redis.call('HSET', KEYS[1], 'discoveryComplete', 1)
            local discovered = tonumber(redis.call('HGET', KEYS[1], 'discovered')) or 0
-            local processed = tonumber(redis.call('HGET', KEYS[1], 'processed')) or 0
+            local articleProcessed = tonumber(redis.call('HGET', KEYS[1], 'articleProcessed')) or 0
+            local deliveryExpected = tonumber(redis.call('HGET', KEYS[1], 'deliveryExpected')) or 0
+            local deliveryProcessed = tonumber(redis.call('HGET', KEYS[1], 'deliveryProcessed')) or 0
            local terminal = 0
-            if processed >= discovered then
+            if articleProcessed >= discovered and deliveryProcessed >= deliveryExpected then
                redis.call('HSET', KEYS[1], 'terminalSent', 1)
                terminal = 1
            end
@@ -314,31 +143,35 @@ impl JobQueue {
        Ok(open == 1)
    }

-    pub async fn record_run_result(
+    pub async fn record_article_result(
        &self,
        run_id: &str,
        job_id: &str,
        persisted: usize,
        delivered: usize,
+        delivery_expected: usize,
        failed: usize,
    ) -> Result<Option<QueuedRunUpdate>> {
        const SCRIPT: &str = r#"
            if redis.call('EXISTS', KEYS[1]) == 0 then return {} end
            if tonumber(redis.call('HGET', KEYS[1], 'terminalSent')) == 1 then return {} end
            if redis.call('HSETNX', KEYS[1], 'job:' .. ARGV[1], 1) == 0 then return {} end
-            local processed = redis.call('HINCRBY', KEYS[1], 'processed', 1)
+            local articleProcessed = redis.call('HINCRBY', KEYS[1], 'articleProcessed', 1)
            local persisted = redis.call('HINCRBY', KEYS[1], 'persisted', ARGV[2])
            local delivered = redis.call('HINCRBY', KEYS[1], 'delivered', ARGV[3])
-            local failed = redis.call('HINCRBY', KEYS[1], 'failed', ARGV[4])
+            local deliveryExpected = redis.call('HINCRBY', KEYS[1], 'deliveryExpected', ARGV[4])
+            local failed = redis.call('HINCRBY', KEYS[1], 'failed', ARGV[5])
            local discovered = tonumber(redis.call('HGET', KEYS[1], 'discovered')) or 0
+            local deliveryProcessed = tonumber(redis.call('HGET', KEYS[1], 'deliveryProcessed')) or 0
            local terminal = 0
            if tonumber(redis.call('HGET', KEYS[1], 'discoveryComplete')) == 1
-                and processed >= discovered
+                and articleProcessed >= discovered
+                and deliveryProcessed >= deliveryExpected
                and tonumber(redis.call('HGET', KEYS[1], 'terminalSent')) == 0 then
                redis.call('HSET', KEYS[1], 'terminalSent', 1)
                terminal = 1
            end
-            redis.call('EXPIRE', KEYS[1], ARGV[5])
+            redis.call('EXPIRE', KEYS[1], ARGV[6])
            return {discovered, persisted, delivered, failed, terminal}
        "#;
        let key = self.run_progress_key(run_id);
@@ -351,6 +184,7 @@ impl JobQueue {
            .arg(job_id)
            .arg(persisted)
            .arg(delivered)
+            .arg(delivery_expected)
            .arg(failed)
            .arg(RUN_PROGRESS_TTL_SECONDS)
            .invoke_async(&mut connection)
@@ -358,18 +192,34 @@ impl JobQueue {
        queued_update_from_values(&values)
    }

-    pub async fn fail_run(&self, run_id: &str) -> Result<Option<RunMetrics>> {
+    pub async fn record_delivery_result(
+        &self,
+        run_id: &str,
+        job_id: &str,
+        delivered: usize,
+        failed: usize,
+    ) -> Result<Option<QueuedRunUpdate>> {
        const SCRIPT: &str = r#"
            if redis.call('EXISTS', KEYS[1]) == 0 then return {} end
            if tonumber(redis.call('HGET', KEYS[1], 'terminalSent')) == 1 then return {} end
-            redis.call('HSET', KEYS[1], 'terminalSent', 1)
-            redis.call('EXPIRE', KEYS[1], ARGV[1])
-            return {
-                tonumber(redis.call('HGET', KEYS[1], 'discovered')) or 0,
-                tonumber(redis.call('HGET', KEYS[1], 'persisted')) or 0,
-                tonumber(redis.call('HGET', KEYS[1], 'delivered')) or 0,
-                tonumber(redis.call('HGET', KEYS[1], 'failed')) or 0
-            }
+            if redis.call('HSETNX', KEYS[1], 'deliveryJob:' .. ARGV[1], 1) == 0 then return {} end
+            local deliveryProcessed = redis.call('HINCRBY', KEYS[1], 'deliveryProcessed', 1)
+            local delivered = redis.call('HINCRBY', KEYS[1], 'delivered', ARGV[2])
+            local failed = redis.call('HINCRBY', KEYS[1], 'failed', ARGV[3])
+            local discovered = tonumber(redis.call('HGET', KEYS[1], 'discovered')) or 0
+            local articleProcessed = tonumber(redis.call('HGET', KEYS[1], 'articleProcessed')) or 0
+            local deliveryExpected = tonumber(redis.call('HGET', KEYS[1], 'deliveryExpected')) or 0
+            local terminal = 0
+            if tonumber(redis.call('HGET', KEYS[1], 'discoveryComplete')) == 1
+                and articleProcessed >= discovered
+                and deliveryProcessed >= deliveryExpected
+                and tonumber(redis.call('HGET', KEYS[1], 'terminalSent')) == 0 then
+                redis.call('HSET', KEYS[1], 'terminalSent', 1)
+                terminal = 1
+            end
+            redis.call('EXPIRE', KEYS[1], ARGV[4])
+            return {discovered, tonumber(redis.call('HGET', KEYS[1], 'persisted')) or 0,
+                delivered, failed, terminal}
        "#;
        let key = self.run_progress_key(run_id);
        let mut connection = self
@@ -378,27 +228,27 @@ impl JobQueue {
            .await?;
        let values: Vec<i64> = redis::Script::new(SCRIPT)
            .key(key)
+            .arg(job_id)
+            .arg(delivered)
+            .arg(failed)
            .arg(RUN_PROGRESS_TTL_SECONDS)
            .invoke_async(&mut connection)
            .await?;
-        if values.len() != 4 {
-            return Ok(None);
-        }
-        Ok(Some(RunMetrics {
-            articles_discovered: values[0].max(0) as usize,
-            articles_persisted: values[1].max(0) as usize,
-            articles_delivered: values[2].max(0) as usize,
-            articles_failed: values[3].max(0) as usize,
-        }))
+        queued_update_from_values(&values)
+    }
+
+    pub async fn fail_run(&self, run_id: &str) -> Result<Option<RunMetrics>> {
+        self.close_run(run_id).await
    }

    pub async fn status(&self) -> Result<(Vec<QueueSnapshot>, Vec<OpenQueuedRun>)> {
-        let (discovery, articles, runs) = tokio::try_join!(
+        let (discovery, articles, delivery, runs) = tokio::try_join!(
            queue_snapshot(&self.discovery, &self.discovery_name),
            queue_snapshot(&self.articles, &self.articles_name),
+            queue_snapshot(&self.delivery, &self.delivery_name),
            self.open_runs(),
        )?;
-        Ok((vec![discovery, articles], runs))
+        Ok((vec![discovery, articles, delivery], runs))
    }

    pub async fn open_runs(&self) -> Result<Vec<OpenQueuedRun>> {
@@ -425,7 +275,9 @@ impl JobQueue {
                        "sourceId",
                        "startedAt",
                        "discovered",
-                        "processed",
+                        "articleProcessed",
+                        "deliveryExpected",
+                        "deliveryProcessed",
                        "persisted",
                        "delivered",
                        "failed",
@@ -433,7 +285,7 @@ impl JobQueue {
                    ])
                    .query_async(&mut connection)
                    .await?;
-                if values.len() != 8 || values[7].as_deref() == Some("1") {
+                if values.len() != 10 || values[9].as_deref() == Some("1") {
                    continue;
                }
                let Some(run_id) = key.rsplit(':').next() else {
@@ -454,9 +306,9 @@ impl JobQueue {
                };
                let metrics = [
                    parse_metric(values[2].as_deref().unwrap_or("0")),
-                    parse_metric(values[4].as_deref().unwrap_or("0")),
-                    parse_metric(values[5].as_deref().unwrap_or("0")),
                    parse_metric(values[6].as_deref().unwrap_or("0")),
+                    parse_metric(values[7].as_deref().unwrap_or("0")),
+                    parse_metric(values[8].as_deref().unwrap_or("0")),
                ];
                let [Ok(discovered), Ok(persisted), Ok(delivered), Ok(failed)] = metrics else {
                    tracing::warn!(run_id, "skipping queued run tracker with invalid metrics");
@@ -469,7 +321,9 @@ impl JobQueue {
                        started_at: started_at.with_timezone(&Utc),
                    },
                    source_id,
-                    processed: parse_metric(values[3].as_deref().unwrap_or("0"))?,
+                    articles_processed: parse_metric(values[3].as_deref().unwrap_or("0"))?,
+                    deliveries_expected: parse_metric(values[4].as_deref().unwrap_or("0"))?,
+                    deliveries_processed: parse_metric(values[5].as_deref().unwrap_or("0"))?,
                    metrics: RunMetrics {
                        articles_discovered: discovered,
                        articles_persisted: persisted,
@@ -500,6 +354,7 @@ impl JobQueue {
    pub async fn reset_agent(&self) -> Result<AgentResetReport> {
        self.discovery.obliterate(true, 1_000).await?;
        self.articles.obliterate(true, 1_000).await?;
+        self.delivery.obliterate(true, 1_000).await?;

        let pattern = self.run_progress_pattern();
        let mut connection = self
@@ -534,6 +389,7 @@ impl JobQueue {
            agent_id: self.agent_id.clone(),
            discovery_queue: self.discovery_name.clone(),
            articles_queue: self.articles_name.clone(),
+            delivery_queue: self.delivery_name.clone(),
            progress_trackers_removed,
        })
    }
@@ -586,117 +442,3 @@ impl JobQueue {
        }))
    }
 }
-
-pub fn queue_options(config: &QueueConfig) -> QueueOptions {
-    QueueOptions::new()
-        .connection(redis_options(config))
-        .prefix(config.prefix.clone())
-}
-
-pub fn redis_options(config: &QueueConfig) -> RedisConnectionOptions {
-    RedisConnectionOptions {
-        url: config.redis_url.clone(),
-        ..RedisConnectionOptions::default()
-    }
-}
-
-async fn queue_snapshot(queue: &Queue, name: &str) -> Result<QueueSnapshot> {
-    let (counts, workers) = tokio::try_join!(queue.get_job_counts(), queue.get_workers_count())?;
-    Ok(snapshot_from_counts(name, workers, counts))
-}
-
-fn snapshot_from_counts(name: &str, workers: usize, counts: JobCounts) -> QueueSnapshot {
-    QueueSnapshot {
-        name: name.to_owned(),
-        workers,
-        waiting: counts.waiting,
-        active: counts.active,
-        delayed: counts.delayed,
-        prioritized: counts.prioritized,
-        completed: counts.completed,
-        failed: counts.failed,
-        waiting_children: counts.waiting_children,
-        paused: counts.paused,
-    }
-}
-
-fn retention(seconds: u64) -> RemoveOnFinish {
-    if seconds == 0 {
-        RemoveOnFinish::Bool(true)
-    } else {
-        RemoveOnFinish::Options(KeepJobs {
-            age: Some(Duration::from_secs(seconds).as_millis() as u64),
-            count: None,
-            limit: Some(1_000),
-        })
-    }
-}
-
-fn stable_job_id(prefix: &str, value: &impl Serialize) -> Result<String> {
-    let bytes = serde_json::to_vec(value)?;
-    Ok(format!("{prefix}-{:x}", md5::compute(bytes)))
-}
-
-fn encode_agent_id(agent_id: &str) -> String {
-    let mut encoded = String::with_capacity(agent_id.len());
-    for byte in agent_id.bytes() {
-        if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'.') {
-            encoded.push(char::from(byte));
-        } else {
-            encoded.push('_');
-            encoded.push_str(&format!("{byte:02x}"));
-        }
-    }
-    if encoded.is_empty() {
-        "agent".to_owned()
-    } else {
-        encoded
-    }
-}
-
-fn scoped_queue_name(agent_scope: &str, queue_name: &str) -> String {
-    format!("{agent_scope}-{queue_name}")
-}
-
-fn parse_metric(value: &str) -> Result<usize> {
-    value.parse::<usize>().map_err(|_| {
-        CrawlError::Queue(format!(
-            "invalid queued run metric '{value}' in Redis progress tracker"
-        ))
-    })
-}
-
-fn metrics_from_values(values: &[i64]) -> Result<RunMetrics> {
-    if values.len() != 4 {
-        return Err(CrawlError::Queue(format!(
-            "expected 4 queued run metrics, received {}",
-            values.len()
-        )));
-    }
-    Ok(RunMetrics {
-        articles_discovered: values[0].max(0) as usize,
-        articles_persisted: values[1].max(0) as usize,
-        articles_delivered: values[2].max(0) as usize,
-        articles_failed: values[3].max(0) as usize,
-    })
-}
-
-fn queued_update_from_values(values: &[i64]) -> Result<Option<QueuedRunUpdate>> {
-    if values.is_empty() {
-        return Ok(None);
-    }
-    if values.len() != 5 {
-        return Err(CrawlError::Queue(format!(
-            "expected 5 queued run update values, received {}",
-            values.len()
-        )));
-    }
-    Ok(Some(QueuedRunUpdate {
-        metrics: metrics_from_values(&values[..4])?,
-        terminal: values[4] == 1,
-    }))
-}
-
-#[cfg(test)]
-#[path = "../../tests/unit/execution/queue.rs"]
-mod tests;
@@ -0,0 +1,124 @@
+use std::time::Duration;
+
+use bullmq::options::RedisConnectionOptions;
+use bullmq::types::{JobCounts, KeepJobs, RemoveOnFinish};
+use bullmq::{Queue, QueueOptions};
+use serde::Serialize;
+
+use crate::{
+    config::QueueConfig,
+    error::{CrawlError, Result},
+    telemetry::RunMetrics,
+};
+
+use super::{QueueSnapshot, QueuedRunUpdate};
+
+pub(super) fn queue_options(config: &QueueConfig) -> QueueOptions {
+    QueueOptions::new()
+        .connection(redis_options(config))
+        .prefix(config.prefix.clone())
+}
+
+pub(crate) fn redis_options(config: &QueueConfig) -> RedisConnectionOptions {
+    RedisConnectionOptions {
+        url: config.redis_url.clone(),
+        ..RedisConnectionOptions::default()
+    }
+}
+
+pub(super) async fn queue_snapshot(queue: &Queue, name: &str) -> Result<QueueSnapshot> {
+    let (counts, workers) = tokio::try_join!(queue.get_job_counts(), queue.get_workers_count())?;
+    Ok(snapshot_from_counts(name, workers, counts))
+}
+
+pub(super) fn snapshot_from_counts(name: &str, workers: usize, counts: JobCounts) -> QueueSnapshot {
+    QueueSnapshot {
+        name: name.to_owned(),
+        workers,
+        waiting: counts.waiting,
+        active: counts.active,
+        delayed: counts.delayed,
+        prioritized: counts.prioritized,
+        completed: counts.completed,
+        failed: counts.failed,
+        waiting_children: counts.waiting_children,
+        paused: counts.paused,
+    }
+}
+
+pub(super) fn retention(seconds: u64) -> RemoveOnFinish {
+    if seconds == 0 {
+        RemoveOnFinish::Bool(true)
+    } else {
+        RemoveOnFinish::Options(KeepJobs {
+            age: Some(Duration::from_secs(seconds).as_millis() as u64),
+            count: None,
+            limit: Some(1_000),
+        })
+    }
+}
+
+pub(super) fn stable_job_id(prefix: &str, value: &impl Serialize) -> Result<String> {
+    let bytes = serde_json::to_vec(value)?;
+    Ok(format!("{prefix}-{:x}", md5::compute(bytes)))
+}
+
+pub(super) fn encode_agent_id(agent_id: &str) -> String {
+    let mut encoded = String::with_capacity(agent_id.len());
+    for byte in agent_id.bytes() {
+        if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'.') {
+            encoded.push(char::from(byte));
+        } else {
+            encoded.push('_');
+            encoded.push_str(&format!("{byte:02x}"));
+        }
+    }
+    if encoded.is_empty() {
+        "agent".to_owned()
+    } else {
+        encoded
+    }
+}
+
+pub(super) fn scoped_queue_name(agent_scope: &str, queue_name: &str) -> String {
+    format!("{agent_scope}-{queue_name}")
+}
+
+pub(super) fn parse_metric(value: &str) -> Result<usize> {
+    value.parse::<usize>().map_err(|_| {
+        CrawlError::Queue(format!(
+            "invalid queued run metric '{value}' in Redis progress tracker"
+        ))
+    })
+}
+
+pub(super) fn metrics_from_values(values: &[i64]) -> Result<RunMetrics> {
+    if values.len() != 4 {
+        return Err(CrawlError::Queue(format!(
+            "expected 4 queued run metrics, received {}",
+            values.len()
+        )));
+    }
+    Ok(RunMetrics {
+        articles_discovered: values[0].max(0) as usize,
+        articles_persisted: values[1].max(0) as usize,
+        articles_delivered: values[2].max(0) as usize,
+        articles_failed: values[3].max(0) as usize,
+    })
+}
+
+pub(super) fn queued_update_from_values(values: &[i64]) -> Result<Option<QueuedRunUpdate>> {
+    if values.is_empty() {
+        return Ok(None);
+    }
+    if values.len() != 5 {
+        return Err(CrawlError::Queue(format!(
+            "expected 5 queued run update values, received {}",
+            values.len()
+        )));
+    }
+    Ok(Some(QueuedRunUpdate {
+        metrics: metrics_from_values(&values[..4])?,
+        terminal: values[4] == 1,
+    }))
+}
@@ -1,5 +1,8 @@
 //! BullMQ worker orchestration.

+mod delivery;
+mod discovery;
+
 use std::sync::Arc;

 use bullmq::worker::WorkerEvent;
@@ -9,13 +12,16 @@ use tokio::sync::Semaphore;
 use tokio::time::{Duration, interval};

 use crate::{
-    articles::{ArticleIngestionClient, IngestStatus, Outbox, ingest},
+    articles::{ArticleIngestionClient, DeliveryIntent, DeliveryStatus, Outbox, normalize},
    error::{CrawlError, Result},
-    execution::{DiscoverJob, FetchJob, JobQueue, Runtime},
+    execution::{DeliveryJob, DiscoverJob, FetchJob, JobQueue, Runtime},
    sources::SourceAdapter,
-    telemetry::{AgentReporter, RunMetrics, RunReporter},
+    telemetry::{AgentReporter, RunReporter},
 };

+use delivery::{enqueue_delivery_intent, process_delivery, reconcile_delivery_intents};
+use discovery::process_discovery;
+
 pub async fn run_worker(
    runtime: Runtime,
    queue_names: Vec<String>,
@@ -31,6 +37,7 @@ pub async fn run_worker(

    let outbox = Outbox::open(&runtime.config.sqlite_path(), true)?;
    let ingestion = ArticleIngestionClient::new(&runtime.config.ingestion, runtime.http.clone())?;
+    reconcile_delivery_intents(&jobs, &outbox).await?;
    let permits = Arc::new(Semaphore::new(concurrency.max(1)));
    let mut workers = Vec::with_capacity(queue_names.len());

@@ -77,6 +84,19 @@ pub async fn run_worker(
            heartbeat_reporter.heartbeat().await;
        }
    });
+    let reconciliation_jobs = jobs.clone();
+    let reconciliation_outbox = outbox.clone();
+    let reconciliation_task = tokio::spawn(async move {
+        let mut ticker = interval(Duration::from_secs(30));
+        loop {
+            ticker.tick().await;
+            if let Err(error) =
+                reconcile_delivery_intents(&reconciliation_jobs, &reconciliation_outbox).await
+            {
+                tracing::warn!(%error, "could not reconcile pending delivery intents");
+            }
+        }
+    });

    tracing::info!(
        agent_id = runtime.agent_id,
@@ -87,6 +107,7 @@ pub async fn run_worker(
    );
    tokio::signal::ctrl_c().await.map_err(CrawlError::Io)?;
    heartbeat_task.abort();
+    reconciliation_task.abort();
    tracing::info!("shutdown requested; draining BullMQ workers");
    let mut close_error = None;
    for worker in &workers {
@@ -182,7 +203,7 @@ async fn process_job(
                );
                return Ok(Value::Null);
            }
-            match process_article(runtime, outbox, ingestion, &payload).await {
+            match process_article(runtime, jobs, outbox, &payload).await {
                Ok(outcome) => {
                    report_article_result(runtime, jobs, &payload, &job_id, outcome).await?;
                    Ok(Value::Null)
@@ -205,112 +226,40 @@ async fn process_job(
                }
            }
        }
+        "deliver-article" => {
+            let job_id = job.id().to_owned();
+            let final_attempt = job.attempts_made() + 1 >= job.opts().attempts.unwrap_or(1);
+            let payload: DeliveryJob = serde_json::from_value(job.data().clone())?;
+            process_delivery(
+                runtime,
+                jobs,
+                outbox,
+                ingestion,
+                &payload,
+                &job_id,
+                final_attempt,
+            )
+            .await?;
+            Ok(Value::Null)
+        }
        name => Err(bullmq::Error::Unrecoverable(format!(
            "unknown crawler job '{name}'"
        ))),
    }
 }

-async fn process_discovery(
-    runtime: &Runtime,
-    jobs: &JobQueue,
-    payload: DiscoverJob,
-    final_attempt: bool,
-) -> Result<usize> {
-    let mut request = payload.request;
-    runtime.config.prepare_request(&mut request)?;
-    let reporter = queued_run_reporter(runtime, &payload.run, &request.source_id);
-    reporter.started().await;
-    let result: Result<usize> = async {
-        runtime.resolve_date_range(&mut request).await;
-        let source = runtime.config.source(&request.source_id)?;
-        let adapter = SourceAdapter::new(source, runtime.http.clone());
-        let mut batches = adapter.stream_discovery(request.clone());
-        let mut count = 0usize;
-        while let Some(batch) = batches.recv().await {
-            let batch = batch?;
-            if !jobs.run_is_open(&payload.run.run_id).await? {
-                tracing::info!(
-                    run_id = payload.run.run_id,
-                    source = %request.source_id,
-                    "queued run was completed during discovery; stopping"
-                );
-                return Ok(count);
-            }
-            let batch_count = batch.articles.len();
-            let Some(metrics) = jobs
-                .record_discovery_batch(&payload.run.run_id, &batch.id, batch_count)
-                .await?
-            else {
-                return Ok(count);
-            };
-            count = metrics.articles_discovered;
-            reporter.progress(metrics).await;
-            for article in batch.articles {
-                jobs.enqueue_article(FetchJob {
-                    request: request.clone(),
-                    article,
-                    run: payload.run.clone(),
-                })
-                .await?;
-            }
-        }
-
-        if let Some(update) = jobs.finish_discovery(&payload.run.run_id).await?
-            && update.terminal
-        {
-            reporter
-                .completed(update.metrics, queued_duration_ms(payload.run.started_at))
-                .await;
-        }
-        tracing::info!(source = %request.source_id, count, "discovery job queued articles");
-        Ok(count)
-    }
-    .await;
-
-    if let Err(error) = &result {
-        if final_attempt {
-            let metrics = match jobs.fail_run(&payload.run.run_id).await {
-                Ok(Some(metrics)) => metrics,
-                Ok(None) => RunMetrics::default(),
-                Err(tracking_error) => {
-                    tracing::warn!(
-                        run_id = payload.run.run_id,
-                        %tracking_error,
-                        "could not close queued run progress tracker"
-                    );
-                    RunMetrics::default()
-                }
-            };
-            reporter
-                .failed(
-                    metrics,
-                    queued_duration_ms(payload.run.started_at),
-                    error.to_string(),
-                )
-                .await;
-        } else {
-            tracing::warn!(
-                run_id = payload.run.run_id,
-                %error,
-                "discovery job attempt failed; BullMQ will retry it"
-            );
-        }
-    }
-    result
-}
-
 #[derive(Debug, Clone, Copy, Default)]
 struct ArticleRunOutcome {
    persisted: usize,
    delivered: usize,
+    delivery_expected: usize,
    failed: usize,
 }

 async fn process_article(
    runtime: &Runtime,
+    jobs: &JobQueue,
    outbox: &Outbox,
-    ingestion: Option<&ArticleIngestionClient>,
    payload: &FetchJob,
 ) -> Result<ArticleRunOutcome> {
    let source = runtime.config.source(&payload.request.source_id)?;
@@ -328,23 +277,41 @@ async fn process_article(
        }
        Err(error) => return Err(error),
    };
-    let (_, status) = ingest(draft, outbox, ingestion).await?;
+    let article = normalize(draft)?;
+    let intent = DeliveryIntent {
+        run_id: payload.run.run_id.clone(),
+        agent_id: payload.run.agent_id.clone(),
+        source_id: payload.request.source_id.clone(),
+        article_hash: article.hash.clone(),
+        started_at: payload.run.started_at,
+    };
+    let status = outbox.save_with_delivery_intent(&article, &intent)?;
    tracing::info!(url = %payload.article.url, ?status, "article job completed");
    Ok(match status {
-        IngestStatus::Persisted => ArticleRunOutcome {
-            persisted: 1,
-            ..ArticleRunOutcome::default()
-        },
-        IngestStatus::AlreadyForwarded | IngestStatus::Forwarded => ArticleRunOutcome {
-            persisted: 1,
-            delivered: 1,
-            failed: 0,
-        },
-        IngestStatus::DeliveryFailed => ArticleRunOutcome {
-            persisted: 1,
-            delivered: 0,
-            failed: 1,
-        },
+        DeliveryStatus::Forwarded => {
+            if outbox.has_delivery_intent(&payload.run.run_id, article.hash.as_str())? {
+                ArticleRunOutcome {
+                    persisted: 1,
+                    delivery_expected: 1,
+                    ..ArticleRunOutcome::default()
+                }
+            } else {
+                ArticleRunOutcome {
+                    persisted: 1,
+                    delivered: 1,
+                    failed: 0,
+                    delivery_expected: 0,
+                }
+            }
+        }
+        DeliveryStatus::Pending | DeliveryStatus::Failed => {
+            enqueue_delivery_intent(jobs, outbox, &intent).await?;
+            ArticleRunOutcome {
+                persisted: 1,
+                delivery_expected: 1,
+                ..ArticleRunOutcome::default()
+            }
+        }
    })
 }

@@ -356,11 +323,12 @@ async fn report_article_result(
    outcome: ArticleRunOutcome,
 ) -> bullmq::Result<()> {
    let update = jobs
-        .record_run_result(
+        .record_article_result(
            &payload.run.run_id,
            job_id,
            outcome.persisted,
            outcome.delivered,
+            outcome.delivery_expected,
            outcome.failed,
        )
        .await
@@ -378,7 +346,7 @@ async fn report_article_result(
    Ok(())
 }

-fn queued_run_reporter(
+pub(super) fn queued_run_reporter(
    runtime: &Runtime,
    run: &super::queue::QueuedRunContext,
    source_id: &crate::domain::SourceId,
@@ -392,13 +360,13 @@ fn queued_run_reporter(
    )
 }

-fn queued_duration_ms(started_at: chrono::DateTime<chrono::Utc>) -> u64 {
+pub(super) fn queued_duration_ms(started_at: chrono::DateTime<chrono::Utc>) -> u64 {
    chrono::Utc::now()
        .signed_duration_since(started_at)
        .num_milliseconds()
        .max(0) as u64
 }

-fn processing_error(error: CrawlError) -> bullmq::Error {
+pub(super) fn processing_error(error: CrawlError) -> bullmq::Error {
    bullmq::Error::ProcessingError(error.to_string())
 }
@@ -0,0 +1,129 @@
+use crate::{
+    articles::{ArticleIngestionClient, DeliveryIntent, DeliveryResult, DeliveryStatus, Outbox},
+    error::Result,
+    execution::{DeliveryJob, JobQueue, QueuedRunContext, Runtime},
+};
+
+use super::{processing_error, queued_duration_ms, queued_run_reporter};
+
+pub(super) async fn process_delivery(
+    runtime: &Runtime,
+    jobs: &JobQueue,
+    outbox: &Outbox,
+    ingestion: Option<&ArticleIngestionClient>,
+    payload: &DeliveryJob,
+    job_id: &str,
+    final_attempt: bool,
+) -> bullmq::Result<()> {
+    let Some(record) = outbox
+        .get(payload.article_hash.as_str())
+        .map_err(processing_error)?
+    else {
+        tracing::error!(
+            article_hash = %payload.article_hash,
+            "delivery article is missing from SQLite"
+        );
+        return finish_delivery(runtime, jobs, outbox, payload, job_id, false).await;
+    };
+
+    if record.status == DeliveryStatus::Forwarded {
+        finish_delivery(runtime, jobs, outbox, payload, job_id, true).await?;
+        return Ok(());
+    }
+
+    let Some(ingestion) = ingestion else {
+        let message = "delivery queue requires an ingestion API endpoint";
+        outbox
+            .mark_failed(payload.article_hash.as_str(), message, false)
+            .map_err(processing_error)?;
+        tracing::error!(article_hash = %payload.article_hash, %message);
+        return finish_delivery(runtime, jobs, outbox, payload, job_id, false).await;
+    };
+    match ingestion.deliver(&record.article).await {
+        DeliveryResult::Delivered { .. } => {
+            outbox
+                .mark_forwarded(payload.article_hash.as_str())
+                .map_err(processing_error)?;
+            finish_delivery(runtime, jobs, outbox, payload, job_id, true).await
+        }
+        DeliveryResult::Failed {
+            retryable, message, ..
+        } => {
+            outbox
+                .mark_failed(payload.article_hash.as_str(), &message, retryable)
+                .map_err(processing_error)?;
+            if !retryable || final_attempt {
+                finish_delivery(runtime, jobs, outbox, payload, job_id, false).await
+            } else {
+                Err(bullmq::Error::ProcessingError(message))
+            }
+        }
+    }
+}
+
+async fn finish_delivery(
+    runtime: &Runtime,
+    jobs: &JobQueue,
+    outbox: &Outbox,
+    payload: &DeliveryJob,
+    job_id: &str,
+    succeeded: bool,
+) -> bullmq::Result<()> {
+    let update = jobs
+        .record_delivery_result(
+            &payload.run.run_id,
+            job_id,
+            usize::from(succeeded),
+            usize::from(!succeeded),
+        )
+        .await
+        .map_err(processing_error)?;
+    outbox
+        .complete_delivery_intent(
+            &payload.run.run_id,
+            payload.article_hash.as_str(),
+            succeeded,
+        )
+        .map_err(processing_error)?;
+    let Some(update) = update else {
+        return Ok(());
+    };
+    let reporter = queued_run_reporter(runtime, &payload.run, &payload.source_id);
+    reporter.progress(update.metrics).await;
+    if update.terminal {
+        reporter
+            .completed(update.metrics, queued_duration_ms(payload.run.started_at))
+            .await;
+    }
+    Ok(())
+}
+
+pub(super) async fn reconcile_delivery_intents(jobs: &JobQueue, outbox: &Outbox) -> Result<usize> {
+    jobs.retry_failed_deliveries().await?;
+    let intents = outbox.pending_delivery_intents(1_000)?;
+    for intent in &intents {
+        enqueue_delivery_intent(jobs, outbox, intent).await?;
+    }
+    if !intents.is_empty() {
+        tracing::info!(count = intents.len(), "reconciled pending delivery intents");
+    }
+    Ok(intents.len())
+}
+
+pub(super) async fn enqueue_delivery_intent(
+    jobs: &JobQueue,
+    outbox: &Outbox,
+    intent: &DeliveryIntent,
+) -> Result<()> {
+    jobs.enqueue_delivery(DeliveryJob {
+        article_hash: intent.article_hash.clone(),
+        source_id: intent.source_id.clone(),
+        run: QueuedRunContext {
+            run_id: intent.run_id.clone(),
+            agent_id: intent.agent_id.clone(),
+            started_at: intent.started_at,
+        },
+    })
+    .await?;
+    outbox.mark_delivery_intent_queued(&intent.run_id, intent.article_hash.as_str())
+}
@@ -0,0 +1,99 @@
+use crate::{
+    error::Result,
+    execution::{DiscoverJob, FetchJob, JobQueue, Runtime},
+    sources::SourceAdapter,
+    telemetry::RunMetrics,
+};
+
+use super::{queued_duration_ms, queued_run_reporter};
+
+pub(super) async fn process_discovery(
+    runtime: &Runtime,
+    jobs: &JobQueue,
+    payload: DiscoverJob,
+    final_attempt: bool,
+) -> Result<usize> {
+    let mut request = payload.request;
+    runtime.config.prepare_request(&mut request)?;
+    let reporter = queued_run_reporter(runtime, &payload.run, &request.source_id);
+    reporter.started().await;
+    let result: Result<usize> = async {
+        runtime.resolve_date_range(&mut request).await;
+        let source = runtime.config.source(&request.source_id)?;
+        let adapter = SourceAdapter::new(source, runtime.http.clone());
+        let mut batches = adapter.stream_discovery(request.clone());
+        let mut count = 0usize;
+        while let Some(batch) = batches.recv().await {
+            let batch = batch?;
+            if !jobs.run_is_open(&payload.run.run_id).await? {
+                tracing::info!(
+                    run_id = payload.run.run_id,
+                    source = %request.source_id,
+                    "queued run was completed during discovery; stopping"
+                );
+                return Ok(count);
+            }
+            let mut article_job_ids = Vec::with_capacity(batch.articles.len());
+            for article in batch.articles {
+                article_job_ids.push(
+                    jobs.enqueue_article(FetchJob {
+                        request: request.clone(),
+                        article,
+                        run: payload.run.clone(),
+                    })
+                    .await?,
+                );
+            }
+            let Some(metrics) = jobs
+                .record_discovery_batch(&payload.run.run_id, &batch.id, &article_job_ids)
+                .await?
+            else {
+                return Ok(count);
+            };
+            count = metrics.articles_discovered;
+            reporter.progress(metrics).await;
+        }
+
+        if let Some(update) = jobs.finish_discovery(&payload.run.run_id).await?
+            && update.terminal
+        {
+            reporter
+                .completed(update.metrics, queued_duration_ms(payload.run.started_at))
+                .await;
+        }
+        tracing::info!(source = %request.source_id, count, "discovery job queued articles");
+        Ok(count)
+    }
+    .await;
+
+    if let Err(error) = &result {
+        if final_attempt {
+            let metrics = match jobs.fail_run(&payload.run.run_id).await {
+                Ok(Some(metrics)) => metrics,
+                Ok(None) => RunMetrics::default(),
+                Err(tracking_error) => {
+                    tracing::warn!(
+                        run_id = payload.run.run_id,
+                        %tracking_error,
+                        "could not close queued run progress tracker"
+                    );
+                    RunMetrics::default()
+                }
+            };
+            reporter
+                .failed(
+                    metrics,
+                    queued_duration_ms(payload.run.started_at),
+                    error.to_string(),
+                )
+                .await;
+        } else {
+            tracing::warn!(
+                run_id = payload.run.run_id,
+                %error,
+                "discovery job attempt failed; BullMQ will retry it"
+            );
+        }
+    }
+    result
+}
@@ -20,7 +20,8 @@ pub use crawler::{
    AgentResetReport, Crawler, CrawlerStatus, OpenRunStatus, QueueStatus, RedisStatus,
 };
 pub use domain::{
-    Article, ArticleDraft, ArticleMetadata, CrawlRequest, DateRange, PageRange, SourceId,
+    AgentId, Article, ArticleDraft, ArticleHash, ArticleMetadata, CategorySlug, CrawlRequest,
+    DateRange, DeliveryOutcome, DeliveryState, PageRange, RetryDecision, RunId, SourceId,
    UpdateDirection,
 };
 pub use error::{CrawlError, Result};
@@ -1,9 +1,11 @@
 //! Generic CSS-selector-driven HTML crawler.

+mod parser;
+
 use std::{collections::HashSet, time::Duration};

 use regex::Regex;
-use scraper::{ElementRef, Html, Selector};
+use scraper::Html;
 use tokio::sync::mpsc;
 use tokio::time::sleep;
 use url::Url;
@@ -16,6 +18,8 @@ use crate::{
    sources::{ArticleSeed, DiscoveryBatch, common},
 };

+use parser::{ListingEntry, element_text, extract_attribute, extract_text, parse_selector};
+
 pub struct HtmlCrawler {
    source: HtmlSourceConfig,
    http: HttpClient,
@@ -393,55 +397,6 @@ impl HtmlCrawler {
    }
 }

-struct ListingEntry {
-    html: String,
-}
-
-fn parse_selector(value: &str) -> Result<Selector> {
-    Selector::parse(value).map_err(|error| {
-        CrawlError::InvalidSourceSelectors(format!("selector '{value}' is invalid: {error}"))
-    })
-}
-
-fn extract_text(document: &Html, selector: &str) -> Result<Option<String>> {
-    let selector = parse_selector(selector)?;
-    Ok(document.select(&selector).next().and_then(|element| {
-        let name = element.value().name();
-        let special = match name {
-            "img" => element
-                .value()
-                .attr("alt")
-                .or_else(|| element.value().attr("title")),
-            "time" => element.value().attr("datetime"),
-            "meta" => element.value().attr("content"),
-            _ => None,
-        };
-        special
-            .map(str::trim)
-            .filter(|value| !value.is_empty())
-            .map(str::to_owned)
-            .or_else(|| element_text(element))
-    }))
-}
-
-fn extract_attribute(document: &Html, selector: &str) -> Result<Option<String>> {
-    let selector = parse_selector(selector)?;
-    Ok(document.select(&selector).next().and_then(|element| {
-        element
-            .value()
-            .attr("href")
-            .or_else(|| element.value().attr("data-href"))
-            .or_else(|| element.value().attr("src"))
-            .map(str::to_owned)
-    }))
-}
-
-fn element_text(element: ElementRef<'_>) -> Option<String> {
-    let value = element.text().collect::<Vec<_>>().join(" ");
-    let value = value.split_whitespace().collect::<Vec<_>>().join(" ");
-    (!value.is_empty()).then_some(value)
-}
-
 #[cfg(test)]
 #[path = "../../tests/unit/sources/html.rs"]
 mod tests;
@@ -0,0 +1,52 @@
+use scraper::{ElementRef, Html, Selector};
+
+use crate::error::{CrawlError, Result};
+
+pub(super) struct ListingEntry {
+    pub(super) html: String,
+}
+
+pub(super) fn parse_selector(value: &str) -> Result<Selector> {
+    Selector::parse(value).map_err(|error| {
+        CrawlError::InvalidSourceSelectors(format!("selector '{value}' is invalid: {error}"))
+    })
+}
+
+pub(super) fn extract_text(document: &Html, selector: &str) -> Result<Option<String>> {
+    let selector = parse_selector(selector)?;
+    Ok(document.select(&selector).next().and_then(|element| {
+        let name = element.value().name();
+        let special = match name {
+            "img" => element
+                .value()
+                .attr("alt")
+                .or_else(|| element.value().attr("title")),
+            "time" => element.value().attr("datetime"),
+            "meta" => element.value().attr("content"),
+            _ => None,
+        };
+        special
+            .map(str::trim)
+            .filter(|value| !value.is_empty())
+            .map(str::to_owned)
+            .or_else(|| element_text(element))
+    }))
+}
+
+pub(super) fn extract_attribute(document: &Html, selector: &str) -> Result<Option<String>> {
+    let selector = parse_selector(selector)?;
+    Ok(document.select(&selector).next().and_then(|element| {
+        element
+            .value()
+            .attr("href")
+            .or_else(|| element.value().attr("data-href"))
+            .or_else(|| element.value().attr("src"))
+            .map(str::to_owned)
+    }))
+}
+
+pub(super) fn element_text(element: ElementRef<'_>) -> Option<String> {
+    let value = element.text().collect::<Vec<_>>().join(" ");
+    let value = value.split_whitespace().collect::<Vec<_>>().join(" ");
+    (!value.is_empty()).then_some(value)
+}
@@ -6,7 +6,7 @@ use super::*;

 fn article() -> Article {
    Article {
-        hash: "hash-1".into(),
+        hash: crate::domain::ArticleHash::new("11111111111111111111111111111111").unwrap(),
        title: "Title".into(),
        body: "Body".into(),
        link: Url::parse("https://example.com/one").unwrap(),
@@ -29,6 +29,64 @@ fn forwarded_rows_stay_forwarded_when_saved_again() {
    assert_eq!(outbox.save(&article).unwrap(), DeliveryStatus::Forwarded);
 }

+#[test]
+fn delivery_intent_is_persisted_with_the_article_and_completed() {
+    let directory = tempdir().unwrap();
+    let path = directory.path().join("outbox.db");
+    let outbox = Outbox::open(&path, true).unwrap();
+    let article = article();
+    let intent = DeliveryIntent {
+        run_id: "019c0000-0000-7000-8000-000000000001".into(),
+        agent_id: "pi-01".into(),
+        source_id: article.source_id.clone(),
+        article_hash: article.hash.clone(),
+        started_at: Utc::now(),
+    };
+
+    assert_eq!(
+        outbox.save_with_delivery_intent(&article, &intent).unwrap(),
+        DeliveryStatus::Pending
+    );
+    let pending = outbox.pending_delivery_intents(10).unwrap();
+    assert_eq!(pending.len(), 1);
+    assert_eq!(pending[0].run_id, intent.run_id);
+    assert_eq!(pending[0].article_hash, article.hash);
+
+    outbox
+        .mark_delivery_intent_queued(&intent.run_id, article.hash.as_str())
+        .unwrap();
+    assert_eq!(outbox.pending_delivery_intents(10).unwrap().len(), 1);
+    assert_eq!(outbox.stats().unwrap().delivery_intents_pending, 1);
+
+    outbox
+        .complete_delivery_intent(&intent.run_id, article.hash.as_str(), true)
+        .unwrap();
+    assert_eq!(outbox.stats().unwrap().delivery_intents_pending, 0);
+}
+
+#[test]
+fn clearing_articles_cascades_to_delivery_intents() {
+    let directory = tempdir().unwrap();
+    let path = directory.path().join("outbox.db");
+    let outbox = Outbox::open(&path, true).unwrap();
+    let article = article();
+    outbox
+        .save_with_delivery_intent(
+            &article,
+            &DeliveryIntent {
+                run_id: "019c0000-0000-7000-8000-000000000002".into(),
+                agent_id: "pi-01".into(),
+                source_id: article.source_id.clone(),
+                article_hash: article.hash.clone(),
+                started_at: Utc::now(),
+            },
+        )
+        .unwrap();
+
+    assert_eq!(outbox.clear().unwrap(), 1);
+    assert_eq!(outbox.stats().unwrap().delivery_intents_pending, 0);
+}
+
 #[test]
 fn claim_reserves_pending_rows() {
    let directory = tempdir().unwrap();
@@ -127,13 +185,13 @@ fn stats_summarize_delivery_and_claim_state() {
        .unwrap();

    let mut second = article();
-    second.hash = "hash-2".into();
+    second.hash = crate::domain::ArticleHash::new("22222222222222222222222222222222").unwrap();
    second.link = Url::parse("https://example.com/two").unwrap();
    outbox.save(&second).unwrap();
    outbox.mark_failed(&second.hash, "temporary", true).unwrap();

    let mut third = article();
-    third.hash = "hash-3".into();
+    third.hash = crate::domain::ArticleHash::new("33333333333333333333333333333333").unwrap();
    third.link = Url::parse("https://example.com/three").unwrap();
    outbox.save(&third).unwrap();
    outbox.mark_forwarded(&third.hash).unwrap();
@@ -147,6 +205,8 @@ fn stats_summarize_delivery_and_claim_state() {
            failed: 1,
            retryable_failed: 1,
            claimed: 1,
+            delivery_intents_pending: 0,
+            delivery_intents_failed: 0,
        }
    );
 }
@@ -5,6 +5,7 @@ fn bundled_configuration_matches_the_zod_and_rust_schemas() {
    let config = loader::parse(loader::BUNDLED_CONFIG).unwrap();
    assert_eq!(config.queue.queues.discovery, "discovery");
    assert_eq!(config.queue.queues.articles, "articles");
+    assert_eq!(config.queue.queues.delivery, "delivery");
    assert!(matches!(config.sources[0], SourceConfig::Html(_)));
 }

@@ -1,4 +1,5 @@
 use super::*;
+use chrono::{DateTime, Utc};

 #[test]
 fn page_range_rejects_reversed_bounds() {