Read the property selection from task.inputs; route takes only task_id 🟩

Per PR review: the FE writes the selection config
({portfolio_id, property_ids?, select_all?}) into the FE-owned task.inputs and
passes only task_id, so a large hand-picked selection never travels in an HTTP
body. The route reads task.inputs, resolves to landlord_property_ids, caps, and
pins the recipe onto sub_task.inputs as before. Declares the FE-owned inputs
column on the TaskRow mirror so the backend can read it and the test schema
builds it.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

backend/app/documents/download_tasks.py

@@ -8,7 +8,7 @@ Modelling Run distributor works around — see ``modelling/run_tasks.py``).
 
 import json
 from datetime import datetime, timezone
-from typing import Any, Optional
+from typing import Any, Optional, cast
 from uuid import UUID
 
 from sqlalchemy import text
@@ -19,6 +19,23 @@ class DocumentDownloadTasks:
     def __init__(self, session: Session) -> None:
         self._session = session
 
+    def read_selection_config(self, task_id: UUID) -> dict[str, Any]:
+        """The task's selection config, read from the FE-owned ``task.inputs``
+        JSON (ADR-0060) — ``{portfolio_id, property_ids?, select_all?}``. Empty
+        dict when the task has no inputs."""
+        row = (
+            self._session.connection()
+            .execute(
+                text("SELECT inputs FROM tasks WHERE id = :task_id"),
+                {"task_id": task_id},
+            )
+            .first()
+        )
+        if row is None or row[0] is None:
+            return {}
+        parsed: Any = json.loads(row[0])
+        return cast(dict[str, Any], parsed) if isinstance(parsed, dict) else {}
+
     def already_distributed(self, task_id: UUID) -> bool:
         """Whether the task already has a sub_task — the 409 guard against a
         double-submit re-triggering the same download."""

backend/app/documents/router.py

@@ -84,21 +84,30 @@ async def bulk_download(
             ),
         )
 
-    if not body.select_all and not body.property_ids:
+    config = tasks.read_selection_config(body.task_id)
+    portfolio_id = config.get("portfolio_id")
+    property_ids = config.get("property_ids")
+    select_all = bool(config.get("select_all", False))
+    if not isinstance(portfolio_id, int):
         raise HTTPException(
             status_code=400,
-            detail="Provide property_ids or set select_all.",
+            detail="task.inputs is missing an integer portfolio_id.",
+        )
+    if not select_all and not property_ids:
+        raise HTTPException(
+            status_code=400,
+            detail="task.inputs must provide property_ids or set select_all.",
         )
 
     property_count, landlord_property_ids = tasks.resolve_selection(
-        body.portfolio_id, body.property_ids, body.select_all
+        portfolio_id, property_ids, select_all
     )
     if property_count == 0:
         raise HTTPException(
             status_code=400,
             detail=(
                 f"The selection resolves to no properties in portfolio "
-                f"{body.portfolio_id}."
+                f"{portfolio_id}."
             ),
         )
     if property_count > MAX_PROPERTIES:
@@ -122,7 +131,7 @@ async def bulk_download(
     recipe = {
         "landlord_property_ids": landlord_property_ids,
         "recipient_email": recipient_email,
-        "package_name": f"portfolio-{body.portfolio_id}-{body.task_id}",
+        "package_name": f"portfolio-{portfolio_id}-{body.task_id}",
     }
     tasks.create_download_subtask(body.task_id, subtask_id, recipe)
 

backend/app/documents/schemas.py

@@ -1,4 +1,3 @@
-from typing import Optional
 from uuid import UUID
 
 from pydantic import BaseModel
@@ -7,13 +6,11 @@ from pydantic import BaseModel
 class BulkDownloadRequest(BaseModel):
     """A request to assemble a Download Package (ADR-0060).
 
-    The front end creates the app-owned Task first (its own request lands in
-    ``task.inputs``) and passes ``task_id`` here. Either an explicit
-    ``property_ids`` selection or ``select_all`` (the whole portfolio) must be
-    given.
+    The front end creates the app-owned Task first and writes the **selection
+    config** into ``task.inputs`` (a JSON object of
+    ``{portfolio_id, property_ids?, select_all?}`` — so a large hand-picked set
+    never travels in an HTTP body), then passes only the ``task_id`` here. The
+    backend reads the selection from ``task.inputs``.
     """
 
     task_id: UUID
-    portfolio_id: int
-    property_ids: Optional[list[int]] = None
-    select_all: bool = False

docs/adr/0060-bulk-document-download-package-model.md

@@ -22,13 +22,20 @@ property, each holding the latest file of each Document Type — built
 best-effort, size-capped at the trigger, on the app-owned-task + attach-mode
 lane (ADR-0055).**
 
-- **Trigger & lifecycle (ADR-0055).** The FastAPI route creates the `tasks`
-  row, resolves the selection to a concrete `property_id` set (expanding "all
-  properties" of a portfolio), **pins that set plus the recipient email into
-  `tasks.inputs`** (TEXT/JSON — a large id list is why this lives in `inputs`,
-  not the 256 KB SQS body), pre-creates **one** `sub_task`, and drops one SQS
-  message (`task_id`, `sub_task_id`). The new `applications/bulk_document_download`
-  Lambda runs in **attach mode**; `TaskOrchestrator` owns status + roll-up.
+- **Trigger & lifecycle (ADR-0055).** The **front end** creates the app-owned
+  `tasks` row and writes the **selection config** —
+  `{portfolio_id, property_ids?, select_all?}` — into `tasks.inputs` (TEXT/JSON,
+  FE-owned), then calls the FastAPI route with **only the `task_id`**. A large
+  hand-picked selection therefore never travels in an HTTP body. The route
+  **reads the selection from `tasks.inputs`**, resolves it to the distinct
+  `landlord_property_id` set, caps it, resolves the recipient email from the
+  authenticated user (ADR-0059), **pins the resolved recipe**
+  (`landlord_property_ids`, `recipient_email`, `package_name`) onto **one
+  pre-created `sub_task`'s `inputs`**, and drops one SQS message (`task_id`,
+  `sub_task_id`). The new `applications/bulk_document_download` Lambda runs in
+  **attach mode**, reading that recipe from the sub_task; `TaskOrchestrator`
+  owns status + roll-up. (`task.inputs` = the FE's selection; `sub_task.inputs`
+  = the backend's resolved recipe.)
 - **One package = one sub_task, streamed.** No fan-out. A Download Package can
   be **several GB**, so it is never held whole in memory: each document is read
   from its own `s3_file_bucket`, written into an on-disk ZIP in `/tmp`, and

infrastructure/postgres/task_table.py

@@ -2,7 +2,7 @@ from datetime import datetime, timezone
 from typing import ClassVar, Optional
 from uuid import UUID, uuid4
 
-from sqlalchemy import Column
+from sqlalchemy import Column, Text
 from sqlalchemy import Enum as SAEnum
 from sqlmodel import Field, SQLModel
 
@@ -18,6 +18,11 @@ class TaskRow(SQLModel, table=True):
     job_completed: Optional[datetime] = None
     status: str = Field(default="waiting")
     service: Optional[str] = None
+    # FE-owned (Drizzle) TEXT column holding the task's request as a JSON
+    # string. Declared here so the backend can read it (e.g. the Bulk Document
+    # Download route reads its property selection from it, ADR-0060) and so the
+    # test schema builds the column; prod owns the real column via FE migrations.
+    inputs: Optional[str] = Field(default=None, sa_column=Column(Text, nullable=True))
     updated_at: datetime = Field(
         default_factory=lambda: datetime.now(timezone.utc)
     )

tests/backend/app/documents/test_bulk_download_router.py

@@ -8,6 +8,7 @@ import json
 from collections.abc import Iterator
 from dataclasses import dataclass, field
 from typing import Any
+from uuid import UUID
 
 import pytest
 from fastapi import FastAPI
@@ -22,10 +23,9 @@ from backend.app.documents.router import (
     get_requesting_user_email,
     get_session,
 )
-from domain.tasks.tasks import Task
 from infrastructure.postgres.property_table import PropertyRow
+from infrastructure.postgres.task_table import TaskRow
 from repositories.tasks.subtask_postgres_repository import SubTaskPostgresRepository
-from repositories.tasks.task_postgres_repository import TaskPostgresRepository
 
 PORTFOLIO_ID = 814
 RECIPIENT = "tester@example.com"
@@ -37,11 +37,18 @@ class Api:
     engine: Engine
     sent_bodies: list[str] = field(default_factory=list)
 
-    def seed_task(self) -> Task:
+    def seed_task(self, selection: dict[str, Any]) -> UUID:
+        """Create the app-owned task with the FE-written selection config on
+        ``task.inputs`` (ADR-0060)."""
         with Session(self.engine) as session:
-            return TaskPostgresRepository(session).create(
-                Task.create(task_source="app:bulk_document_download")
+            row = TaskRow(
+                task_source="app:bulk_document_download",
+                inputs=json.dumps(selection),
             )
+            session.add(row)
+            session.commit()
+            session.refresh(row)
+            return row.id
 
     def seed_property(self, landlord_property_id: str) -> int:
         with Session(self.engine) as session:
@@ -54,9 +61,9 @@ class Api:
             assert row.id is not None
             return row.id
 
-    def subtask_inputs(self, task: Task) -> list[dict[str, Any]]:
+    def subtask_inputs(self, task_id: UUID) -> list[dict[str, Any]]:
         with Session(self.engine) as session:
-            subtasks = SubTaskPostgresRepository(session).list_by_task(task.id)
+            subtasks = SubTaskPostgresRepository(session).list_by_task(task_id)
             return [st.inputs or {} for st in subtasks]
 
 
@@ -77,64 +84,76 @@ def api(db_engine: Engine) -> Api:
     return harness
 
 
-def _trigger(api: Api, task: Task, property_ids: list[int]) -> Any:
+def _trigger(api: Api, task_id: UUID) -> Any:
     return api.client.post(
-        "/v1/documents/bulk-download",
-        json={
-            "task_id": str(task.id),
-            "portfolio_id": PORTFOLIO_ID,
-            "property_ids": property_ids,
-        },
+        "/v1/documents/bulk-download", json={"task_id": str(task_id)}
     )
 
 
 def test_pins_the_recipe_and_enqueues_one_message(api: Api) -> None:
-    # arrange — two selected properties, matched by landlord_property_id.
-    task = api.seed_task()
+    # arrange — two selected properties, pinned as the FE's task.inputs config.
     p1 = api.seed_property("LP1")
     p2 = api.seed_property("LP2")
+    task_id = api.seed_task(
+        {"portfolio_id": PORTFOLIO_ID, "property_ids": [p1, p2]}
+    )
 
     # act
-    response = _trigger(api, task, [p1, p2])
+    response = _trigger(api, task_id)
 
     # assert — accepted; one sub_task pins the recipe; one message enqueued.
     assert response.status_code == 202
-    inputs = api.subtask_inputs(task)
+    inputs = api.subtask_inputs(task_id)
     assert len(inputs) == 1
     assert inputs[0]["landlord_property_ids"] == ["LP1", "LP2"]
     assert inputs[0]["recipient_email"] == RECIPIENT
     assert len(api.sent_bodies) == 1
     message = json.loads(api.sent_bodies[0])
-    assert message["task_id"] == str(task.id)
+    assert message["task_id"] == str(task_id)
     assert "subtask_id" in message
 
 
 def test_rejects_a_selection_over_the_cap(api: Api, monkeypatch: Any) -> None:
     # arrange — cap of 1, two selected properties.
     monkeypatch.setattr(documents_router, "MAX_PROPERTIES", 1)
-    task = api.seed_task()
     p1 = api.seed_property("LP1")
     p2 = api.seed_property("LP2")
+    task_id = api.seed_task(
+        {"portfolio_id": PORTFOLIO_ID, "property_ids": [p1, p2]}
+    )
 
     # act
-    response = _trigger(api, task, [p1, p2])
+    response = _trigger(api, task_id)
 
     # assert — refused; nothing created or sent.
     assert response.status_code == 400
-    assert api.subtask_inputs(task) == []
+    assert api.subtask_inputs(task_id) == []
     assert api.sent_bodies == []
 
 
 def test_rejects_a_task_that_already_has_a_subtask(api: Api) -> None:
     # arrange — a task already triggered once.
-    task = api.seed_task()
     p1 = api.seed_property("LP1")
-    assert _trigger(api, task, [p1]).status_code == 202
+    task_id = api.seed_task({"portfolio_id": PORTFOLIO_ID, "property_ids": [p1]})
+    assert _trigger(api, task_id).status_code == 202
 
     # act
-    response = _trigger(api, task, [p1])
+    response = _trigger(api, task_id)
 
     # assert — the double-submit is refused; still one sub_task, one message.
     assert response.status_code == 409
-    assert len(api.subtask_inputs(task)) == 1
+    assert len(api.subtask_inputs(task_id)) == 1
     assert len(api.sent_bodies) == 1
+
+
+def test_rejects_a_task_with_no_selection_config(api: Api) -> None:
+    # arrange — a task whose inputs carry no portfolio_id/selection.
+    api.seed_property("LP1")
+    task_id = api.seed_task({})
+
+    # act
+    response = _trigger(api, task_id)
+
+    # assert
+    assert response.status_code == 400
+    assert api.sent_bodies == []