CUST-WP-0074-T02: task wait qualifiers in the hub read model

Adds tasks.decision_id and workplan_dependencies.from_task_id (migration
e8f9a0b1c2d3). A from_task_id column is needed rather than
from_workplan_id plus description: the partial unique indexes key on
(from_workplan_id, target, relationship_type), so two tasks in one
workplan waiting on the same target could not both be indexed, and the
derived wait_kind needs per-task attribution. The two existing unique
indexes are narrowed to frontmatter edges (from_task_id IS NULL) and two
task-origin counterparts are added; downgrade deletes task-origin rows
before restoring the old indexes.

POST /workplans/{id}/dependencies/ accepts from_task_id (must belong to
the from workplan; a task cannot depend on itself). TaskCreate/Update/Read
carry decision_id.

Derived read-model fields, no new write routes (rule 5):
- TaskRead.wait_kind: external | human | both | unqualified, null unless
  status is wait (dependency rows from this task / needs_human).
- WorkplanRead.blocked_kind: human | external | none, null unless status
  is blocked. Human wins over external.
Both come from api/services/wait_kind.py, applied on GET /tasks/,
GET /tasks/{id}, GET /workplans/ and GET /workplans/{id}.

The identifier-migration reference-count proof now includes from_task_id
(FK count guard 22 -> 23).

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

Assistant: claude-code
Assistant-Model: sonnet
Assistant-Process: 237582@bnt-lap001
Assistant-Session: f2b3d9f1-8fb9-4b9c-bc2b-837ec5dfc826

tests/test_wait_kind.py

@@ -0,0 +1,126 @@
+"""CUST-WP-0074-T02: derived wait_kind / blocked_kind read-model fields and the
+from_task_id dependency edge. Real PostgreSQL via the shared `client` fixture."""
+from __future__ import annotations
+
+import pytest
+
+from api.services.wait_kind import derive_blocked_kind, derive_wait_kind
+from tests.conftest import create_test_domain, create_test_repo, create_test_workplan
+
+
+class TestDeriveFunctions:
+    def test_wait_kind_by_qualifiers(self):
+        assert derive_wait_kind("todo", needs_human=True, has_dependency=True) is None
+        assert derive_wait_kind("wait", needs_human=False, has_dependency=False) == "unqualified"
+        assert derive_wait_kind("wait", needs_human=False, has_dependency=True) == "external"
+        assert derive_wait_kind("wait", needs_human=True, has_dependency=False) == "human"
+        assert derive_wait_kind("wait", needs_human=True, has_dependency=True) == "both"
+
+    def test_blocked_kind_prefers_human(self):
+        assert derive_blocked_kind("active", ["human"]) is None
+        assert derive_blocked_kind("blocked", []) == "none"
+        assert derive_blocked_kind("blocked", ["unqualified", None]) == "none"
+        assert derive_blocked_kind("blocked", ["external", "unqualified"]) == "external"
+        assert derive_blocked_kind("blocked", ["external", "human"]) == "human"
+        assert derive_blocked_kind("blocked", ["both"]) == "human"
+
+
+async def _task(client, workplan_id, title, **extra):
+    r = await client.post("/tasks/", json={"workplan_id": workplan_id, "title": title, **extra})
+    assert r.status_code == 201, r.text
+    return r.json()
+
+
+@pytest.mark.asyncio
+async def test_wait_kind_and_blocked_kind_on_read_routes(client):
+    domain = await create_test_domain(client)
+    repo = await create_test_repo(client, domain_slug=domain["slug"])
+    upstream = await create_test_workplan(client, repo["id"], slug="upstream", status="active")
+    blocked = await create_test_workplan(client, repo["id"], slug="blocked", status="blocked")
+
+    human = await _task(
+        client, blocked["id"], "human gate", status="wait",
+        needs_human=True, blocking_reason="operator must provision creds",
+        intervention_note="operator must provision creds", decision_id="CCR-2026-0004",
+    )
+    external = await _task(
+        client, blocked["id"], "external", status="wait", blocking_reason="upstream first",
+    )
+    unqualified = await _task(client, blocked["id"], "unqualified", status="wait", blocking_reason="?")
+    todo = await _task(client, blocked["id"], "plain todo")
+
+    r = await client.post(
+        f"/workplans/{blocked['id']}/dependencies/",
+        json={"from_task_id": external["id"], "to_workplan_id": upstream["id"]},
+    )
+    assert r.status_code == 201, r.text
+    assert r.json()["from_task_id"] == external["id"]
+
+    r = await client.get("/tasks/", params={"workplan_id": blocked["id"]})
+    assert r.status_code == 200
+    kinds = {row["title"]: row["wait_kind"] for row in r.json()}
+    assert kinds == {
+        "human gate": "human", "external": "external", "unqualified": "unqualified", "plain todo": None,
+    }
+    r = await client.get(f"/tasks/{human['id']}")
+    assert r.json()["wait_kind"] == "human"
+    assert r.json()["decision_id"] == "CCR-2026-0004"
+
+    r = await client.get(f"/workplans/{blocked['id']}")
+    assert r.json()["blocked_kind"] == "human"
+    r = await client.get(f"/workplans/{upstream['id']}")
+    assert r.json()["blocked_kind"] is None
+
+    # Human gate lifted → the workplan is blocked on the external wait only.
+    r = await client.patch(f"/tasks/{human['id']}", json={"status": "done"})
+    assert r.status_code == 200, r.text
+    r = await client.get("/workplans/", params={"repo_id": repo["id"]})
+    kinds = {row["slug"]: row["blocked_kind"] for row in r.json()}
+    assert kinds == {"upstream": None, "blocked": "external"}
+
+    # Dependency and human flag together → both.
+    r = await client.patch(
+        f"/tasks/{external['id']}",
+        json={"needs_human": True, "intervention_note": "confirm with owner"},
+    )
+    assert r.status_code == 200, r.text
+    r = await client.get(f"/tasks/{external['id']}")
+    assert r.json()["wait_kind"] == "both"
+    del unqualified, todo
+
+
+@pytest.mark.asyncio
+async def test_from_task_must_belong_to_the_from_workplan(client):
+    domain = await create_test_domain(client)
+    repo = await create_test_repo(client, domain_slug=domain["slug"])
+    a = await create_test_workplan(client, repo["id"], slug="a")
+    b = await create_test_workplan(client, repo["id"], slug="b")
+    task_in_b = await _task(client, b["id"], "in b")
+    r = await client.post(
+        f"/workplans/{a['id']}/dependencies/",
+        json={"from_task_id": task_in_b["id"], "to_workplan_id": b["id"]},
+    )
+    assert r.status_code == 404
+    r = await client.post(
+        f"/workplans/{b['id']}/dependencies/",
+        json={"from_task_id": task_in_b["id"], "to_task_id": task_in_b["id"]},
+    )
+    assert r.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_two_tasks_may_wait_on_the_same_target(client):
+    domain = await create_test_domain(client)
+    repo = await create_test_repo(client, domain_slug=domain["slug"])
+    upstream = await create_test_workplan(client, repo["id"], slug="upstream")
+    wp = await create_test_workplan(client, repo["id"], slug="waiting")
+    t1 = await _task(client, wp["id"], "t1")
+    t2 = await _task(client, wp["id"], "t2")
+    for task in (t1, t2):
+        r = await client.post(
+            f"/workplans/{wp['id']}/dependencies/",
+            json={"from_task_id": task["id"], "to_workplan_id": upstream["id"]},
+        )
+        assert r.status_code == 201, r.text
+    r = await client.get(f"/workplans/{wp['id']}/dependencies/")
+    assert sorted(row["from_task_id"] for row in r.json()) == sorted([t1["id"], t2["id"]])

tests/test_work_record_identifier_migration.py

@@ -188,7 +188,7 @@ def test_every_work_record_foreign_key_cascades_on_update():
         for foreign_key in table.foreign_keys
         if foreign_key.target_fullname in {"workplans.id", "tasks.id"}
     ]
-    assert len(references) == 22
+    assert len(references) == 23
     assert all(foreign_key.onupdate == "CASCADE" for foreign_key in references)