diff --git a/digital_land/pipeline/__init__.py b/digital_land/pipeline/__init__.py
index 0936fc54..61845963 100644
--- a/digital_land/pipeline/__init__.py
+++ b/digital_land/pipeline/__init__.py
@@ -7,3 +7,4 @@
     run_pipeline,
     EntityNumGen,
 )
+from .task import TaskPipeline, TaskPipelineConfig, TaskPipelineStatus  # noqa: F401
diff --git a/digital_land/pipeline/task.py b/digital_land/pipeline/task.py
new file mode 100644
index 00000000..67b8f13b
--- /dev/null
+++ b/digital_land/pipeline/task.py
@@ -0,0 +1,294 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+from datetime import date
+from enum import Enum
+from pathlib import Path
+from typing import List, Optional
+
+import polars as pl
+from pydantic import ConfigDict, Field
+from pydantic.dataclasses import dataclass
+
+logger = logging.getLogger(__name__)
+
+TASK_COLUMNS = [
+    "dataset",
+    "organisation",
+    "endpoint",
+    "resource",
+    "details",
+    "severity",
+    "responsibility",
+    "task-source",
+    "entry-date",
+    "reference",
+]
+
+_EMPTY_SCHEMA = {col: pl.Utf8 for col in TASK_COLUMNS}
+
+
+class TaskPipelineStatus(Enum):
+    INITIALISED = 1
+    COMPLETE = 2
+    FAILED = 3
+
+
+@dataclass(config=ConfigDict(extra="forbid"))
+class TaskPipelineConfig:
+    dataset: str
+    organisation: str
+    endpoint: str
+    output_path: Path
+    log_path: Optional[Path] = None
+    issue_path: Optional[Path] = None
+    severity_filter: List[str] = Field(default_factory=lambda: ["error"])
+    responsibility_filter: List[str] = Field(default_factory=lambda: ["external"])
+    entry_date: Optional[str] = None
+
+
+class TaskPipeline:
+    """Generates a task CSV from collection logs and/or issue logs."""
+
+    def __init__(self, config: Optional[TaskPipelineConfig] = None):
+        self.config = config
+        self.status = TaskPipelineStatus.INITIALISED
+
+    def run(
+        self,
+        output_path: Path = None,
+        dataset: Optional[str] = None,
+        organisation: Optional[str] = None,
+        endpoint: Optional[str] = None,
+        log_path: Optional[Path] = None,
+        issue_path: Optional[Path] = None,
+        severity_filter: Optional[List[str]] = None,
+        responsibility_filter: Optional[List[str]] = None,
+        entry_date: Optional[str] = None,
+    ) -> TaskPipelineStatus:
+        try:
+            cfg = self.config
+            output_path = output_path or (cfg.output_path if cfg else None)
+            dataset = dataset or (cfg.dataset if cfg else None)
+            organisation = organisation or (cfg.organisation if cfg else None)
+            endpoint = endpoint or (cfg.endpoint if cfg else None)
+            log_path = log_path or (cfg.log_path if cfg else None)
+            issue_path = issue_path or (cfg.issue_path if cfg else None)
+            severity_filter = severity_filter or (
+                cfg.severity_filter if cfg else ["error"]
+            )
+            responsibility_filter = responsibility_filter or (
+                cfg.responsibility_filter if cfg else ["external"]
+            )
+            entry_date = entry_date or (cfg.entry_date if cfg else str(date.today()))
+
+            if output_path is None:
+                logger.error("output_path is required")
+                return TaskPipelineStatus.FAILED
+
+            frames = []
+
+            if log_path and Path(log_path).exists():
+                log_df = (
+                    # infer_schema_length=0 means that polas will interpret all values as strings
+                    # rather than the defaul of infering datatypes on first 100 rows
+                    pl.scan_csv(
+                        log_path, infer_schema_length=0, null_values=[""]
+                    ).collect()
+                )
+                log_tasks = _transform_log_to_tasks(
+                    log_df, dataset, organisation, entry_date
+                )
+                if not log_tasks.is_empty():
+                    frames.append(log_tasks)
+
+            if issue_path and Path(issue_path).exists():
+                issue_df = pl.scan_csv(
+                    issue_path, infer_schema_length=0, null_values=[""]
+                ).collect()
+                issue_tasks = _transform_issues_to_tasks(
+                    issue_df,
+                    organisation,
+                    endpoint,
+                    severity_filter,
+                    responsibility_filter,
+                    entry_date,
+                )
+                if not issue_tasks.is_empty():
+                    frames.append(issue_tasks)
+
+            result = pl.concat(frames) if frames else pl.DataFrame(schema=_EMPTY_SCHEMA)
+            result.write_csv(output_path)
+
+            self.status = TaskPipelineStatus.COMPLETE
+            return TaskPipelineStatus.COMPLETE
+
+        except Exception:
+            logger.exception("TaskPipeline failed")
+            self.status = TaskPipelineStatus.FAILED
+            return TaskPipelineStatus.FAILED
+
+
+def _transform_log_to_tasks(
+    df: pl.DataFrame,
+    dataset: str,
+    organisation: str,
+    entry_date: str,
+) -> pl.DataFrame:
+    """Filter and transform a log DataFrame into task rows."""
+
+    failed = df.filter(pl.col("status") != "200")
+
+    if failed.is_empty():
+        return pl.DataFrame(schema=_EMPTY_SCHEMA)
+
+    n = len(failed)
+
+    details_col = failed.select(
+        # pl.struct bundles multiple columns so map_elements can access both at once
+        pl.struct(["status", "exception"])
+        .map_elements(
+            lambda row: json.dumps(
+                {
+                    # status codes are converted to ints and nulls come in as empty strings
+                    "status": (
+                        int(row["status"])
+                        if row["status"] and row["status"].isdigit()
+                        else row["status"]
+                    ),
+                    "exception": row["exception"] or "",
+                }
+            ),
+            return_dtype=pl.Utf8,
+        )
+        .alias("details")
+    )
+
+    result = pl.DataFrame(
+        {
+            "dataset": pl.Series([dataset] * n, dtype=pl.Utf8),
+            "organisation": pl.Series([organisation] * n, dtype=pl.Utf8),
+            "endpoint": failed["endpoint"],
+            "resource": failed["resource"],
+            "details": details_col["details"],
+            "severity": pl.Series(["error"] * n, dtype=pl.Utf8),
+            "responsibility": pl.Series(["external"] * n, dtype=pl.Utf8),
+            "task-source": pl.Series(["log"] * n, dtype=pl.Utf8),
+            "entry-date": pl.Series([entry_date] * n, dtype=pl.Utf8),
+        }
+    )
+    # calculate a sha256 hash from some key columns to identify this task
+    result = result.with_columns(
+        pl.struct(["dataset", "endpoint", "resource", "task-source", "details"])
+        .map_elements(
+            lambda r: hashlib.sha256(
+                "|".join(
+                    [
+                        r["dataset"] or "",
+                        r["endpoint"] or "",
+                        r["resource"] or "",
+                        r["task-source"],
+                        r["details"],
+                    ]
+                ).encode()
+            ).hexdigest()[:16],
+            return_dtype=pl.Utf8,
+        )
+        .alias("reference")
+    )
+    return result
+
+
+def _transform_issues_to_tasks(
+    df: pl.DataFrame,
+    organisation: str,
+    endpoint: str,
+    severity_filter: List[str],
+    responsibility_filter: List[str],
+    entry_date: str,
+) -> pl.DataFrame:
+    """Filter and transform an issue DataFrame into task rows."""
+
+    cols = set(df.columns)
+
+    if "severity" in cols:
+        df = df.filter(pl.col("severity").is_in(severity_filter))
+    if "responsibility" in cols:
+        df = df.filter(pl.col("responsibility").is_in(responsibility_filter))
+
+    if df.is_empty():
+        return pl.DataFrame(schema=_EMPTY_SCHEMA)
+
+    for col in [
+        "issue-type",
+        "resource",
+        "field",
+        "dataset",
+        "severity",
+        "responsibility",
+    ]:
+        if col not in df.columns:
+            df = df.with_columns(pl.lit("").alias(col))
+
+    # We are counting every row, so there might be duplicate entities across multiple data sources
+    # I can add in a split so it behaves more like the current version in submit repo but holding off
+    # until I understand the data better and if that split is required.
+    grouped = df.group_by(["issue-type", "resource", "field", "dataset"]).agg(
+        [
+            pl.len().alias("count"),
+            pl.first("severity"),
+            pl.first("responsibility"),
+        ]
+    )
+
+    grouped = grouped.with_columns(
+        # pl.struct bundles multiple columns so map_elements can access both at once
+        pl.struct(["issue-type", "count", "field"])
+        .map_elements(
+            lambda row: json.dumps(
+                {
+                    "issue_type": row["issue-type"] or "",
+                    "count": row["count"],
+                    "field": row["field"] or "",
+                }
+            ),
+            return_dtype=pl.Utf8,
+        )
+        .alias("details")
+    )
+
+    result = grouped.select(
+        [
+            pl.col("dataset"),
+            pl.lit(organisation).alias("organisation"),
+            pl.lit(endpoint).alias("endpoint"),
+            pl.col("resource"),
+            pl.col("details"),
+            pl.col("severity"),
+            pl.col("responsibility"),
+            pl.lit("issue").alias("task-source"),
+            pl.lit(entry_date).alias("entry-date"),
+        ]
+    )
+    # calculate a sha256 hash from some key columns to identify this task
+    result = result.with_columns(
+        pl.struct(["dataset", "endpoint", "resource", "task-source", "details"])
+        .map_elements(
+            lambda r: hashlib.sha256(
+                "|".join(
+                    [
+                        r["dataset"] or "",
+                        r["endpoint"] or "",
+                        r["resource"] or "",
+                        r["task-source"],
+                        r["details"],
+                    ]
+                ).encode()
+            ).hexdigest()[:16],
+            return_dtype=pl.Utf8,
+        )
+        .alias("reference")
+    )
+    return result
diff --git a/tests/unit/pipeline/__init__.py b/tests/unit/pipeline/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/unit/pipeline/test_task.py b/tests/unit/pipeline/test_task.py
new file mode 100644
index 00000000..9e9ef50c
--- /dev/null
+++ b/tests/unit/pipeline/test_task.py
@@ -0,0 +1,178 @@
+import csv
+import json
+import pytest
+
+from digital_land.pipeline.task import (
+    TaskPipeline,
+    TaskPipelineConfig,
+    TaskPipelineStatus,
+)
+
+
+@pytest.fixture
+def log_csv(tmp_path):
+    """A collection log with one success and two failures."""
+    path = tmp_path / "log.csv"
+    path.write_text(
+        "endpoint,resource,status,exception,entry-date,bytes,elapsed\n"
+        "endpoint-aaa,resource-aaa,200,,2024-01-01,1234,0.5\n"
+        "endpoint-bbb,resource-bbb,404,,2024-01-01,0,0.1\n"
+        "endpoint-ccc,,500,Connection refused,2024-01-01,0,0.0\n"
+    )
+    return path
+
+
+@pytest.fixture
+def issue_csv(tmp_path):
+    """An issue log with error/external issues and one warning to be filtered out."""
+    path = tmp_path / "issue.csv"
+    path.write_text(
+        "dataset,resource,line-number,entry-number,field,entity,issue-type,value,message,severity,responsibility\n"
+        "tree-preservation-zone,resource-aaa,1,1,geometry,,invalid-geometry,bad-wkt,Invalid geometry,error,external\n"
+        "tree-preservation-zone,resource-aaa,2,2,geometry,,invalid-geometry,other-bad-wkt,Invalid geometry,error,external\n"
+        "tree-preservation-zone,resource-aaa,3,3,name,,missing-value,,Missing value,error,external\n"
+        "tree-preservation-zone,resource-aaa,4,4,notes,,missing-value,,Missing value,warning,internal\n"
+    )
+    return path
+
+
+def _read_csv(path):
+    with open(path) as f:
+        return list(csv.DictReader(f))
+
+
+class TestTaskPipeline:
+
+    def test_run_log_tasks_creates_row_per_failure(self, log_csv, tmp_path):
+        output = tmp_path / "tasks.csv"
+        status = TaskPipeline().run(
+            output_path=output,
+            dataset="tree-preservation-zone",
+            organisation="local-authority-eng:ABC",
+            endpoint="endpoint-aaa",
+            log_path=log_csv,
+        )
+
+        assert status == TaskPipelineStatus.COMPLETE
+        rows = _read_csv(output)
+        assert len(rows) == 2
+        assert set(row["task-source"] for row in rows) == {"log"}
+
+    def test_run_log_task_details_json(self, log_csv, tmp_path):
+        output = tmp_path / "tasks.csv"
+        TaskPipeline().run(
+            output_path=output,
+            dataset="tree-preservation-zone",
+            organisation="local-authority-eng:ABC",
+            endpoint="endpoint-aaa",
+            log_path=log_csv,
+        )
+
+        rows = _read_csv(output)
+        details = [json.loads(row["details"]) for row in rows]
+        statuses = {detail["status"] for detail in details}
+        assert 404 in statuses
+        assert 500 in statuses
+        assert all(isinstance(d["status"], int) for d in details)
+
+    def test_run_log_tasks_empty_when_all_successful(self, tmp_path):
+        log = tmp_path / "log.csv"
+        log.write_text(
+            "endpoint,resource,status,exception,entry-date,bytes,elapsed\n"
+            "endpoint-aaa,resource-aaa,200,,2024-01-01,1234,0.5\n"
+        )
+        output = tmp_path / "tasks.csv"
+        status = TaskPipeline().run(
+            output_path=output,
+            dataset="tree-preservation-zone",
+            organisation="local-authority-eng:ABC",
+            endpoint="endpoint-aaa",
+            log_path=log,
+        )
+        assert status == TaskPipelineStatus.COMPLETE
+        assert _read_csv(output) == []
+
+    def test_run_issue_tasks_groups_by_type_and_field(self, issue_csv, tmp_path):
+        output = tmp_path / "tasks.csv"
+        status = TaskPipeline().run(
+            output_path=output,
+            dataset="tree-preservation-zone",
+            organisation="local-authority-eng:ABC",
+            endpoint="endpoint-aaa",
+            issue_path=issue_csv,
+        )
+
+        assert status == TaskPipelineStatus.COMPLETE
+        rows = _read_csv(output)
+        assert len(rows) == 2
+        assert set(row["task-source"] for row in rows) == {"issue"}
+
+    def test_run_issue_task_details_json(self, issue_csv, tmp_path):
+        output = tmp_path / "tasks.csv"
+        TaskPipeline().run(
+            output_path=output,
+            dataset="tree-preservation-zone",
+            organisation="local-authority-eng:ABC",
+            endpoint="endpoint-aaa",
+            issue_path=issue_csv,
+        )
+
+        rows = _read_csv(output)
+        details = [json.loads(row["details"]) for row in rows]
+        geom_task = next(d for d in details if d["issue_type"] == "invalid-geometry")
+        assert geom_task["count"] == 2
+        assert geom_task["field"] == "geometry"
+
+    def test_run_both_sources_combined(self, log_csv, issue_csv, tmp_path):
+        output = tmp_path / "tasks.csv"
+        status = TaskPipeline().run(
+            output_path=output,
+            dataset="tree-preservation-zone",
+            organisation="local-authority-eng:ABC",
+            endpoint="endpoint-aaa",
+            log_path=log_csv,
+            issue_path=issue_csv,
+        )
+
+        assert status == TaskPipelineStatus.COMPLETE
+        rows = _read_csv(output)
+        sources = set(row["task-source"] for row in rows)
+        assert "log" in sources
+        assert "issue" in sources
+
+    def test_run_with_config(self, log_csv, tmp_path):
+        output = tmp_path / "tasks.csv"
+        config = TaskPipelineConfig(
+            dataset="tree-preservation-zone",
+            organisation="local-authority-eng:ABC",
+            endpoint="endpoint-aaa",
+            output_path=output,
+            log_path=log_csv,
+        )
+        status = TaskPipeline(config).run()
+        assert status == TaskPipelineStatus.COMPLETE
+        assert len(_read_csv(output)) == 2
+
+    def test_run_output_has_correct_columns(self, log_csv, tmp_path):
+        output = tmp_path / "tasks.csv"
+        TaskPipeline().run(
+            output_path=output,
+            dataset="tree-preservation-zone",
+            organisation="local-authority-eng:ABC",
+            endpoint="endpoint-aaa",
+            log_path=log_csv,
+        )
+        expected_cols = {
+            "dataset",
+            "organisation",
+            "endpoint",
+            "resource",
+            "details",
+            "severity",
+            "responsibility",
+            "task-source",
+            "entry-date",
+            "reference",
+        }
+        rows = _read_csv(output)
+        assert set(rows[0].keys()) == expected_cols