CUST-WP-0074-T02 follow-up: blocked_on / wait_until reach the hub read model

Migration f9a0b1c2d3e4 (revises e8f9a0b1c2d3) adds tasks.blocked_on
(String 200) and tasks.wait_until (Date); TaskCreate/Update/Read carry both.

Read model (api/services/wait_kind.py): a wait task is external when it has
a dependency edge, a blocked_on matching ^message-from: (rule 6), or a
wait_until date (a window clears by itself). human/both/unqualified and
blocked_kind are unchanged in shape.

fix-consistency: _task_wait_fields carries blocked_on and wait_until (ISO
date, PyYAML date objects normalised), so the C-06/C-11 creates and the
C-10 status PATCH send them, and C-41 compares and syncs them like the
other qualifiers. C-39 accepts wait_until: YYYY-MM-DD as a qualifier; C-40
also fires when wait_until is before UTC today (report only).

/state/deps: depends_on stubs gain from_task_id / from_task_title so
consumers see task-scoped edges; both optional, response otherwise as before.

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_consistency_check.py

@@ -2364,6 +2364,8 @@ class TestTaskWaitQualifiers:
                 "needs_human": True,
                 "blocking_reason": "operator must provision creds",
                 "decision_id": "CCR-2026-0004",
+                "blocked_on": None,
+                "wait_until": None,
                 "intervention_note": "operator must provision creds",
             },
         }
@@ -2373,6 +2375,73 @@ class TestTaskWaitQualifiers:
         self._install(monkeypatch, self._fake_get(repo, db_task=db_task))
         assert self._issues(check_repo("http://unused", "demo-repo"), "C-41") == []
 
+    def test_c41_syncs_blocked_on_and_wait_until(self, tmp_path, monkeypatch):
+        repo = self._make_repo(
+            tmp_path,
+            "id: STATE-WP-0001-T01\nstatus: wait\npriority: high\n"
+            "state_hub_task_id: \"task-1\"\nblocked_on: message-from:fin-hub\n"
+            "wait_until: 2099-01-31\nblocking_reason: \"fin-hub owner returns an id\"",
+        )
+        db_task = {"id": "task-1", "status": "wait", "needs_human": False,
+                   "blocking_reason": "fin-hub owner returns an id", "decision_id": None,
+                   "blocked_on": None, "wait_until": None}
+        self._install(monkeypatch, self._fake_get(repo, db_task=db_task))
+
+        report = check_repo("http://unused", "demo-repo")
+
+        c41 = self._issues(report, "C-41")
+        assert len(c41) == 1 and c41[0].fixable is True
+        assert c41[0]._fix_context["wait_fields"] == {
+            "needs_human": False,
+            "blocking_reason": "fin-hub owner returns an id",
+            "decision_id": None,
+            "blocked_on": "message-from:fin-hub",
+            "wait_until": "2099-01-31",
+        }
+
+        # The hub echoes the date as ISO text; in sync → silent.
+        db_task.update(blocked_on="message-from:fin-hub", wait_until="2099-01-31")
+        self._install(monkeypatch, self._fake_get(repo, db_task=db_task))
+        assert self._issues(check_repo("http://unused", "demo-repo"), "C-41") == []
+
+    def test_wait_until_qualifies_c39_and_past_date_warns_c40(self, tmp_path, monkeypatch):
+        repo = self._make_repo(
+            tmp_path,
+            "id: STATE-WP-0001-T01\nstatus: wait\npriority: high\n"
+            "state_hub_task_id: \"task-1\"\nwait_until: 2099-01-31\n"
+            "blocking_reason: \"embargo until the window opens\"",
+        )
+        self._install(monkeypatch, self._fake_get(repo))
+        report = check_repo("http://unused", "demo-repo")
+        assert self._issues(report, "C-39") == []
+        assert self._issues(report, "C-40") == []
+
+        repo = self._make_repo(
+            tmp_path / "past",
+            "id: STATE-WP-0001-T01\nstatus: wait\npriority: high\n"
+            "state_hub_task_id: \"task-1\"\nwait_until: \"2020-01-31\"\n"
+            "blocking_reason: \"embargo until the window opens\"",
+        )
+        self._install(monkeypatch, self._fake_get(repo))
+        report = check_repo("http://unused", "demo-repo")
+        assert self._issues(report, "C-39") == []
+        c40 = self._issues(report, "C-40")
+        assert len(c40) == 1 and c40[0].fixable is False
+        assert c40[0].message.startswith("blocker satisfied, task still wait: 'STATE-WP-0001-T01'")
+        assert "wait_until 2020-01-31 has passed" in c40[0].message
+
+        # Not a YYYY-MM-DD date → no qualifier, C-39 as before.
+        repo = self._make_repo(
+            tmp_path / "junk",
+            "id: STATE-WP-0001-T01\nstatus: wait\npriority: high\n"
+            "state_hub_task_id: \"task-1\"\nwait_until: soon\n"
+            "blocking_reason: \"embargo until the window opens\"",
+        )
+        self._install(monkeypatch, self._fake_get(repo))
+        report = check_repo("http://unused", "demo-repo")
+        assert len(self._issues(report, "C-39")) == 1
+        assert self._issues(report, "C-40") == []
+
     def test_c41_ignores_blocks_that_do_not_carry_the_keys(self, tmp_path, monkeypatch):
         repo = self._make_repo(
             tmp_path,
@@ -2431,6 +2500,8 @@ class TestTaskWaitQualifiers:
             "needs_human": False,
             "blocking_reason": "access-engine rename must land first",
             "decision_id": None,
+            "blocked_on": None,
+            "wait_until": None,
         }
         assert ("/tasks/task-1", {"status": "wait", "suppress_token_event": True, **wait_fields}) in patches
         assert ("/tasks/task-1", wait_fields) in patches

tests/test_wait_kind.py

@@ -2,6 +2,8 @@
 from_task_id dependency edge. Real PostgreSQL via the shared `client` fixture."""
 from __future__ import annotations
 
+from datetime import date
+
 import pytest
 
 from api.services.wait_kind import derive_blocked_kind, derive_wait_kind
@@ -16,6 +18,23 @@ class TestDeriveFunctions:
         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_owner_return_marker_and_wait_window_are_external(self):
+        assert derive_wait_kind(
+            "wait", needs_human=False, has_dependency=False, blocked_on="message-from:fin-hub"
+        ) == "external"
+        assert derive_wait_kind(
+            "wait", needs_human=False, has_dependency=False, blocked_on="fin-hub"
+        ) == "unqualified"
+        assert derive_wait_kind(
+            "wait", needs_human=False, has_dependency=False, wait_until=date(2099, 1, 31)
+        ) == "external"
+        assert derive_wait_kind(
+            "wait", needs_human=True, has_dependency=False, wait_until=date(2020, 1, 31)
+        ) == "both"
+        assert derive_wait_kind(
+            "todo", needs_human=False, has_dependency=False, blocked_on="message-from:x"
+        ) is None
+
     def test_blocked_kind_prefers_human(self):
         assert derive_blocked_kind("active", ["human"]) is None
         assert derive_blocked_kind("blocked", []) == "none"
@@ -124,3 +143,74 @@ async def test_two_tasks_may_wait_on_the_same_target(client):
         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"]])
+
+
+@pytest.mark.asyncio
+async def test_blocked_on_and_wait_until_reach_the_read_model(client):
+    domain = await create_test_domain(client)
+    repo = await create_test_repo(client, domain_slug=domain["slug"])
+    blocked = await create_test_workplan(client, repo["id"], slug="blocked", status="blocked")
+
+    owner_return = await _task(
+        client, blocked["id"], "owner return", status="wait",
+        blocking_reason="fin-hub owner returns an id", blocked_on="message-from:fin-hub",
+    )
+    window = await _task(
+        client, blocked["id"], "window", status="wait",
+        blocking_reason="embargo", wait_until="2099-01-31",
+    )
+    plain = await _task(
+        client, blocked["id"], "plain", status="wait", blocking_reason="?", blocked_on="fin-hub",
+    )
+
+    r = await client.get("/tasks/", params={"workplan_id": blocked["id"]})
+    rows = {row["title"]: row for row in r.json()}
+    assert rows["owner return"]["wait_kind"] == "external"
+    assert rows["owner return"]["blocked_on"] == "message-from:fin-hub"
+    assert rows["window"]["wait_kind"] == "external"
+    assert rows["window"]["wait_until"] == "2099-01-31"
+    assert rows["plain"]["wait_kind"] == "unqualified"
+    r = await client.get(f"/workplans/{blocked['id']}")
+    assert r.json()["blocked_kind"] == "external"
+
+    # Clearing the window on PATCH drops the qualifier again.
+    r = await client.patch(f"/tasks/{window['id']}", json={"wait_until": None})
+    assert r.status_code == 200, r.text
+    assert r.json()["wait_until"] is None
+    r = await client.get(f"/tasks/{window['id']}")
+    assert r.json()["wait_kind"] == "unqualified"
+    del owner_return, plain
+
+
+@pytest.mark.asyncio
+async def test_state_deps_stubs_carry_the_waiting_task(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")
+    waiting = await create_test_workplan(client, repo["id"], slug="waiting", status="blocked")
+    upstream_task = await _task(client, upstream["id"], "upstream task")
+    waiter = await _task(client, waiting["id"], "waiter", status="wait", blocking_reason="upstream")
+
+    for body in (
+        {"from_task_id": waiter["id"], "to_workplan_id": upstream["id"]},
+        {"from_task_id": waiter["id"], "to_task_id": upstream_task["id"]},
+        {"to_workplan_id": upstream["id"], "relationship_type": "related"},
+    ):
+        r = await client.post(f"/workplans/{waiting['id']}/dependencies/", json=body)
+        assert r.status_code == 201, r.text
+
+    deps = {row["id"]: row for row in (await client.get("/state/deps")).json()}
+    stubs = {
+        (stub["target_type"], stub["relationship_type"]): stub
+        for stub in deps[waiting["id"]]["depends_on"]
+    }
+    assert stubs[("workstream", "blocks")]["from_task_id"] == waiter["id"]
+    assert stubs[("workstream", "blocks")]["from_task_title"] == "waiter"
+    assert stubs[("task", "blocks")]["from_task_id"] == waiter["id"]
+    assert stubs[("task", "blocks")]["task_id"] == upstream_task["id"]
+    assert stubs[("workstream", "related")]["from_task_id"] is None
+    assert stubs[("workstream", "related")]["from_task_title"] is None
+    # The reverse side is unchanged: blocks stubs name the workplan only.
+    for stub in deps[upstream["id"]]["blocks"]:
+        assert stub["workstream_id"] == waiting["id"]
+        assert stub["from_task_id"] is None