kin: auto-commit after pipeline

core/models.py

@@ -13,7 +13,7 @@ from typing import Any
 
 VALID_TASK_STATUSES = [
     "pending", "in_progress", "review", "done",
-    "blocked", "decomposed", "cancelled",
+    "blocked", "decomposed", "cancelled", "revising",
 ]
 
 VALID_COMPLETION_MODES = {"auto_complete", "review"}
@@ -266,6 +266,48 @@ def get_task(conn: sqlite3.Connection, id: str) -> dict | None:
     return _row_to_dict(row)
 
 
+def get_children(conn: sqlite3.Connection, task_id: str) -> list[dict]:
+    """Return direct child tasks of given task_id."""
+    rows = conn.execute(
+        "SELECT * FROM tasks WHERE parent_task_id = ?", (task_id,)
+    ).fetchall()
+    return _rows_to_list(rows)
+
+
+def has_open_children(conn: sqlite3.Connection, task_id: str) -> bool:
+    """Recursively check if task has any open (not done/cancelled) descendants."""
+    children = get_children(conn, task_id)
+    for child in children:
+        if child["status"] not in ("done", "cancelled"):
+            return True
+        if has_open_children(conn, child["id"]):
+            return True
+    return False
+
+
+def _check_parent_completion(conn: sqlite3.Connection, task_id: str) -> None:
+    """Cascade-check upward: if parent is 'revising' and all children closed → promote to 'done'."""
+    task = get_task(conn, task_id)
+    if not task:
+        return
+    parent_id = task.get("parent_task_id")
+    if not parent_id:
+        return
+    parent = get_task(conn, parent_id)
+    if not parent:
+        return
+    if parent["status"] != "revising":
+        return
+    if not has_open_children(conn, parent_id):
+        now = datetime.now().isoformat()
+        conn.execute(
+            "UPDATE tasks SET status = 'done', completed_at = ?, updated_at = ? WHERE id = ?",
+            (now, now, parent_id),
+        )
+        conn.commit()
+        _check_parent_completion(conn, parent_id)
+
+
 VALID_TASK_SORT_FIELDS = frozenset({
     "updated_at", "created_at", "priority", "status", "title", "id",
 })
@@ -275,14 +317,16 @@ def list_tasks(
     conn: sqlite3.Connection,
     project_id: str | None = None,
     status: str | None = None,
+    parent_task_id: str | None = None,
     limit: int | None = None,
     sort: str = "updated_at",
     sort_dir: str = "desc",
 ) -> list[dict]:
-    """List tasks with optional project/status filters, limit, and sort.
+    """List tasks with optional project/status/parent filters, limit, and sort.
 
     sort: column name (validated against VALID_TASK_SORT_FIELDS, default 'updated_at')
     sort_dir: 'asc' or 'desc' (default 'desc')
+    parent_task_id: filter by parent task id (pass '__none__' to get root tasks)
     """
     # Validate sort field to prevent SQL injection
     sort_col = sort if sort in VALID_TASK_SORT_FIELDS else "updated_at"
@@ -296,6 +340,11 @@ def list_tasks(
     if status:
         query += " AND status = ?"
         params.append(status)
+    if parent_task_id == "__none__":
+        query += " AND parent_task_id IS NULL"
+    elif parent_task_id:
+        query += " AND parent_task_id = ?"
+        params.append(parent_task_id)
     query += f" ORDER BY {sort_col} {sort_direction}"
     if limit is not None:
         query += " LIMIT ?"
@@ -304,7 +353,11 @@ def list_tasks(
 
 
 def update_task(conn: sqlite3.Connection, id: str, **fields) -> dict:
-    """Update task fields. Auto-sets updated_at. Sets completed_at when status transitions to 'done'."""
+    """Update task fields. Auto-sets updated_at. Sets completed_at when status transitions to 'done'.
+
+    If transitioning to 'done' but task has open children, sets 'revising' instead.
+    After a child closes (done/cancelled), cascades upward to promote revising parents.
+    """
     if not fields:
         return get_task(conn, id)
     json_cols = ("brief", "spec", "review", "test_result", "security_result", "labels")
@@ -312,12 +365,20 @@ def update_task(conn: sqlite3.Connection, id: str, **fields) -> dict:
         if key in fields:
             fields[key] = _json_encode(fields[key])
     if "status" in fields and fields["status"] == "done":
-        fields["completed_at"] = datetime.now().isoformat()
+        if has_open_children(conn, id):
+            fields["status"] = "revising"
+            # Do NOT set completed_at for revising (Decision #737)
+        else:
+            fields["completed_at"] = datetime.now().isoformat()
     fields["updated_at"] = datetime.now().isoformat()
     sets = ", ".join(f"{k} = ?" for k in fields)
     vals = list(fields.values()) + [id]
     conn.execute(f"UPDATE tasks SET {sets} WHERE id = ?", vals)
     conn.commit()
+    # Cascade upward: if this child just closed, maybe its parent can finish
+    new_status = fields.get("status")
+    if new_status in ("done", "cancelled"):
+        _check_parent_completion(conn, id)
     return get_task(conn, id)
 
 

tests/test_revising_status.py

@@ -0,0 +1,329 @@
+"""Tests for KIN-127: 'revising' status and recursive child completion blocking.
+
+Covers:
+- VALID_TASK_STATUSES contains 'revising'
+- update_task → done redirects to 'revising' when open children exist
+- Cascade: child→done triggers parent revising→done
+- completed_at semantics for revising vs real done
+- get_children() and has_open_children()
+- API: PATCH status=done, GET /children, GET ?parent_task_id
+"""
+
+import pytest
+from core.db import init_db
+from core import models
+from unittest.mock import patch
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def conn():
+    """Fresh in-memory DB for each test."""
+    c = init_db(db_path=":memory:")
+    yield c
+    c.close()
+
+
+@pytest.fixture
+def proj(conn):
+    """DB connection with a seeded project."""
+    models.create_project(conn, "p1", "P1", "/p1")
+    return conn
+
+
+# ---------------------------------------------------------------------------
+# API client fixture (mirrors test_api.py pattern)
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def client(tmp_path):
+    import web.api as api_module
+    api_module.DB_PATH = tmp_path / "test.db"
+    from web.api import app
+    from fastapi.testclient import TestClient
+    return TestClient(app)
+
+
+# ===========================================================================
+# 1. VALID_TASK_STATUSES
+# ===========================================================================
+
+def test_revising_in_valid_statuses():
+    """'revising' должен присутствовать в VALID_TASK_STATUSES."""
+    assert "revising" in models.VALID_TASK_STATUSES
+
+
+# ===========================================================================
+# 2. update_task: done → revising when open children exist
+# ===========================================================================
+
+def test_done_with_open_children_becomes_revising(proj):
+    """Попытка parent→done при открытом child возвращает status='revising'."""
+    models.create_task(proj, "P1-001", "p1", "Parent")
+    models.create_task(proj, "P1-001a", "p1", "Child", parent_task_id="P1-001", status="pending")
+
+    result = models.update_task(proj, "P1-001", status="done")
+
+    assert result["status"] == "revising"
+
+
+def test_done_without_children_stays_done(proj):
+    """Задача без детей переходит в 'done' без изменений."""
+    models.create_task(proj, "P1-001", "p1", "Task")
+
+    result = models.update_task(proj, "P1-001", status="done")
+
+    assert result["status"] == "done"
+
+
+def test_done_with_all_children_done(proj):
+    """Все дети завершены → parent→done проходит как обычно."""
+    models.create_task(proj, "P1-001", "p1", "Parent")
+    models.create_task(proj, "P1-001a", "p1", "Child A", parent_task_id="P1-001", status="done")
+    models.create_task(proj, "P1-001b", "p1", "Child B", parent_task_id="P1-001", status="done")
+
+    result = models.update_task(proj, "P1-001", status="done")
+
+    assert result["status"] == "done"
+
+
+def test_done_with_cancelled_children(proj):
+    """Дети со статусом 'cancelled' считаются закрытыми → parent→done разрешён."""
+    models.create_task(proj, "P1-001", "p1", "Parent")
+    models.create_task(proj, "P1-001a", "p1", "Child A", parent_task_id="P1-001", status="cancelled")
+    models.create_task(proj, "P1-001b", "p1", "Child B", parent_task_id="P1-001", status="done")
+
+    result = models.update_task(proj, "P1-001", status="done")
+
+    assert result["status"] == "done"
+
+
+# ===========================================================================
+# 3. Cascade: child→done → parent revising→done
+# ===========================================================================
+
+def test_cascade_parent_completion(proj):
+    """child→done автоматически переводит parent из 'revising' в 'done'."""
+    models.create_task(proj, "P1-001", "p1", "Parent")
+    models.create_task(proj, "P1-001a", "p1", "Child", parent_task_id="P1-001", status="pending")
+
+    # parent→done с открытым ребёнком → revising
+    models.update_task(proj, "P1-001", status="done")
+    assert models.get_task(proj, "P1-001")["status"] == "revising"
+
+    # child→done → каскад поднимает parent в done
+    models.update_task(proj, "P1-001a", status="done")
+    assert models.get_task(proj, "P1-001")["status"] == "done"
+
+
+def test_deep_cascade(proj):
+    """3 уровня: grandchild→done → child (revising)→done → parent (revising)→done."""
+    models.create_task(proj, "P1-001", "p1", "Parent")
+    models.create_task(proj, "P1-001a", "p1", "Child", parent_task_id="P1-001", status="pending")
+    models.create_task(proj, "P1-001a1", "p1", "Grandchild", parent_task_id="P1-001a", status="pending")
+
+    # parent→done → revising (есть открытый child, у child открытый grandchild)
+    models.update_task(proj, "P1-001", status="done")
+    assert models.get_task(proj, "P1-001")["status"] == "revising"
+
+    # child→done → revising (есть открытый grandchild)
+    models.update_task(proj, "P1-001a", status="done")
+    assert models.get_task(proj, "P1-001a")["status"] == "revising"
+    # parent остаётся revising
+    assert models.get_task(proj, "P1-001")["status"] == "revising"
+
+    # grandchild→done → cascades: child→done, parent→done
+    models.update_task(proj, "P1-001a1", status="done")
+    assert models.get_task(proj, "P1-001a1")["status"] == "done"
+    assert models.get_task(proj, "P1-001a")["status"] == "done"
+    assert models.get_task(proj, "P1-001")["status"] == "done"
+
+
+# ===========================================================================
+# 4. completed_at semantics
+# ===========================================================================
+
+def test_completed_at_not_set_on_revising(proj):
+    """При переходе в 'revising' completed_at должен оставаться NULL."""
+    models.create_task(proj, "P1-001", "p1", "Parent")
+    models.create_task(proj, "P1-001a", "p1", "Child", parent_task_id="P1-001", status="pending")
+
+    result = models.update_task(proj, "P1-001", status="done")
+
+    assert result["status"] == "revising"
+    assert result["completed_at"] is None
+
+
+def test_completed_at_set_on_real_done(proj):
+    """При каскадном переходе parent из revising → done ставится completed_at."""
+    models.create_task(proj, "P1-001", "p1", "Parent")
+    models.create_task(proj, "P1-001a", "p1", "Child", parent_task_id="P1-001", status="pending")
+
+    models.update_task(proj, "P1-001", status="done")  # → revising
+    models.update_task(proj, "P1-001a", status="done")  # cascade → parent done
+
+    parent = models.get_task(proj, "P1-001")
+    assert parent["status"] == "done"
+    assert parent["completed_at"] is not None
+
+
+# ===========================================================================
+# 5. get_children()
+# ===========================================================================
+
+def test_get_children_returns_direct_children(proj):
+    """get_children возвращает только прямых потомков."""
+    models.create_task(proj, "P1-001", "p1", "Parent")
+    models.create_task(proj, "P1-001a", "p1", "Child A", parent_task_id="P1-001")
+    models.create_task(proj, "P1-001b", "p1", "Child B", parent_task_id="P1-001")
+    models.create_task(proj, "P1-001a1", "p1", "Grandchild", parent_task_id="P1-001a")
+
+    children = models.get_children(proj, "P1-001")
+
+    assert len(children) == 2
+    ids = {c["id"] for c in children}
+    assert ids == {"P1-001a", "P1-001b"}
+
+
+def test_get_children_empty_for_leaf_task(proj):
+    """get_children возвращает [] для задачи без детей."""
+    models.create_task(proj, "P1-001", "p1", "Task")
+
+    assert models.get_children(proj, "P1-001") == []
+
+
+# ===========================================================================
+# 6. has_open_children() — recursive
+# ===========================================================================
+
+def test_has_open_children_recursive(proj):
+    """has_open_children рекурсивно проверяет через 3 уровня вложенности."""
+    models.create_task(proj, "P1-001", "p1", "Parent")
+    models.create_task(proj, "P1-001a", "p1", "Child", parent_task_id="P1-001", status="done")
+    models.create_task(proj, "P1-001a1", "p1", "Grandchild", parent_task_id="P1-001a", status="pending")
+
+    # Child сам done, но у него открытый grandchild — должен вернуть True
+    assert models.has_open_children(proj, "P1-001") is True
+
+
+def test_has_open_children_false_when_all_closed(proj):
+    """has_open_children возвращает False когда все потомки closed."""
+    models.create_task(proj, "P1-001", "p1", "Parent")
+    models.create_task(proj, "P1-001a", "p1", "Child", parent_task_id="P1-001", status="done")
+    models.create_task(proj, "P1-001a1", "p1", "Grandchild", parent_task_id="P1-001a", status="cancelled")
+
+    assert models.has_open_children(proj, "P1-001") is False
+
+
+# ===========================================================================
+# 7. list_tasks — parent_task_id filter
+# ===========================================================================
+
+def test_list_tasks_parent_filter(proj):
+    """list_tasks(parent_task_id=...) возвращает только дочерние задачи."""
+    models.create_task(proj, "P1-001", "p1", "Parent")
+    models.create_task(proj, "P1-001a", "p1", "Child A", parent_task_id="P1-001")
+    models.create_task(proj, "P1-001b", "p1", "Child B", parent_task_id="P1-001")
+    models.create_task(proj, "P1-002", "p1", "Other root task")
+
+    result = models.list_tasks(proj, parent_task_id="P1-001")
+
+    assert len(result) == 2
+    ids = {t["id"] for t in result}
+    assert ids == {"P1-001a", "P1-001b"}
+
+
+def test_list_tasks_parent_none_filter(proj):
+    """list_tasks(parent_task_id='__none__') возвращает только корневые задачи."""
+    models.create_task(proj, "P1-001", "p1", "Root Task")
+    models.create_task(proj, "P1-001a", "p1", "Child", parent_task_id="P1-001")
+
+    result = models.list_tasks(proj, parent_task_id="__none__")
+
+    ids = {t["id"] for t in result}
+    assert "P1-001" in ids
+    assert "P1-001a" not in ids
+
+
+# ===========================================================================
+# 8. API tests
+# ===========================================================================
+
+def test_patch_task_revising_status(client, tmp_path):
+    """PATCH /api/tasks/{id} status=done с открытым child → ответ содержит status='revising'."""
+    import web.api as api_module
+    from core.db import init_db
+    from core import models as m
+
+    client.post("/api/projects", json={"id": "p1", "name": "P1", "path": "/p1"})
+    r = client.post("/api/tasks", json={"project_id": "p1", "title": "Parent Task"})
+    parent_id = r.json()["id"]
+
+    # Seed child directly via models (TaskCreate API has no parent_task_id field)
+    conn = init_db(api_module.DB_PATH)
+    m.create_task(conn, parent_id + "a", "p1", "Child Task",
+                  parent_task_id=parent_id, status="pending")
+    conn.close()
+
+    r = client.patch(f"/api/tasks/{parent_id}", json={"status": "done"})
+
+    assert r.status_code == 200
+    assert r.json()["status"] == "revising"
+
+
+def test_get_children_endpoint(client):
+    """GET /api/tasks/{id}/children возвращает список прямых дочерних задач."""
+    import web.api as api_module
+    from core.db import init_db
+    from core import models as m
+
+    client.post("/api/projects", json={"id": "p1", "name": "P1", "path": "/p1"})
+    r = client.post("/api/tasks", json={"project_id": "p1", "title": "Parent Task"})
+    parent_id = r.json()["id"]
+
+    # Seed children via models
+    conn = init_db(api_module.DB_PATH)
+    m.create_task(conn, parent_id + "a", "p1", "Child A", parent_task_id=parent_id)
+    m.create_task(conn, parent_id + "b", "p1", "Child B", parent_task_id=parent_id)
+    conn.close()
+
+    r = client.get(f"/api/tasks/{parent_id}/children")
+
+    assert r.status_code == 200
+    children = r.json()
+    assert len(children) == 2
+    titles = {c["title"] for c in children}
+    assert titles == {"Child A", "Child B"}
+
+
+def test_get_children_endpoint_not_found(client):
+    """GET /api/tasks/{id}/children для несуществующей задачи → 404."""
+    r = client.get("/api/tasks/NONEXISTENT/children")
+    assert r.status_code == 404
+
+
+def test_list_tasks_parent_filter_api(client):
+    """GET /api/tasks?parent_task_id={id} фильтрует задачи по родителю."""
+    import web.api as api_module
+    from core.db import init_db
+    from core import models as m
+
+    client.post("/api/projects", json={"id": "p1", "name": "P1", "path": "/p1"})
+    r = client.post("/api/tasks", json={"project_id": "p1", "title": "Parent Task"})
+    parent_id = r.json()["id"]
+    client.post("/api/tasks", json={"project_id": "p1", "title": "Other Root Task"})
+
+    # Seed child via models
+    conn = init_db(api_module.DB_PATH)
+    m.create_task(conn, parent_id + "a", "p1", "Child Task", parent_task_id=parent_id)
+    conn.close()
+
+    r = client.get(f"/api/tasks?parent_task_id={parent_id}")
+
+    assert r.status_code == 200
+    tasks = r.json()
+    assert len(tasks) == 1
+    assert tasks[0]["title"] == "Child Task"

tests/test_revising_status_regression.py

@@ -0,0 +1,141 @@
+"""Regression tests for KIN-127 bug: parent_task_id silently ignored in POST /api/tasks.
+
+Root cause: TaskCreate schema was missing parent_task_id field, so child tasks
+created via API had no parent link in DB. Fixed by adding parent_task_id to TaskCreate
+and passing it to models.create_task.
+
+These tests use the API end-to-end (no models bypass) to prevent regression.
+"""
+
+import pytest
+
+
+@pytest.fixture
+def client(tmp_path):
+    import web.api as api_module
+    api_module.DB_PATH = tmp_path / "test.db"
+    from web.api import app
+    from fastapi.testclient import TestClient
+    c = TestClient(app)
+    c.post("/api/projects", json={"id": "p1", "name": "P1", "path": "/p1"})
+    return c
+
+
+# ---------------------------------------------------------------------------
+# 1. parent_task_id сохраняется при создании задачи через API
+# ---------------------------------------------------------------------------
+
+def test_create_task_with_parent_task_id_saves_link(client):
+    """POST /api/tasks с parent_task_id — ссылка на родителя сохраняется в БД."""
+    r = client.post("/api/tasks", json={"project_id": "p1", "title": "Parent"})
+    assert r.status_code == 200
+    parent_id = r.json()["id"]
+
+    r = client.post("/api/tasks", json={
+        "project_id": "p1",
+        "title": "Child",
+        "parent_task_id": parent_id,
+    })
+    assert r.status_code == 200
+    child = r.json()
+    assert child["parent_task_id"] == parent_id
+
+
+def test_create_task_without_parent_task_id_has_no_parent(client):
+    """POST /api/tasks без parent_task_id — задача создаётся как корневая."""
+    r = client.post("/api/tasks", json={"project_id": "p1", "title": "Root task"})
+    assert r.status_code == 200
+    task = r.json()
+    assert task.get("parent_task_id") is None
+
+
+# ---------------------------------------------------------------------------
+# 2. /children возвращает задачи, созданные через API с parent_task_id
+# ---------------------------------------------------------------------------
+
+def test_children_endpoint_returns_api_created_children(client):
+    """GET /children видит дочерние задачи, созданные через POST /api/tasks."""
+    r = client.post("/api/tasks", json={"project_id": "p1", "title": "Parent"})
+    parent_id = r.json()["id"]
+
+    client.post("/api/tasks", json={
+        "project_id": "p1", "title": "Child A", "parent_task_id": parent_id,
+    })
+    client.post("/api/tasks", json={
+        "project_id": "p1", "title": "Child B", "parent_task_id": parent_id,
+    })
+
+    r = client.get(f"/api/tasks/{parent_id}/children")
+    assert r.status_code == 200
+    titles = {c["title"] for c in r.json()}
+    assert titles == {"Child A", "Child B"}
+
+
+def test_children_endpoint_empty_for_task_created_without_parent(client):
+    """GET /children возвращает [] для задачи без дочерних задач."""
+    r = client.post("/api/tasks", json={"project_id": "p1", "title": "Leaf"})
+    task_id = r.json()["id"]
+
+    r = client.get(f"/api/tasks/{task_id}/children")
+    assert r.status_code == 200
+    assert r.json() == []
+
+
+# ---------------------------------------------------------------------------
+# 3. Фильтр ?parent_task_id работает с API-созданными дочерними задачами
+# ---------------------------------------------------------------------------
+
+def test_list_tasks_parent_filter_with_api_created_children(client):
+    """GET /api/tasks?parent_task_id={id} находит задачи, созданные через API."""
+    r = client.post("/api/tasks", json={"project_id": "p1", "title": "Parent"})
+    parent_id = r.json()["id"]
+
+    client.post("/api/tasks", json={
+        "project_id": "p1", "title": "Child", "parent_task_id": parent_id,
+    })
+    client.post("/api/tasks", json={"project_id": "p1", "title": "Other root"})
+
+    r = client.get(f"/api/tasks?parent_task_id={parent_id}")
+    assert r.status_code == 200
+    tasks = r.json()
+    assert len(tasks) == 1
+    assert tasks[0]["title"] == "Child"
+
+
+# ---------------------------------------------------------------------------
+# 4. End-to-end: PATCH status=done → revising, дочерняя задача создана через API
+# ---------------------------------------------------------------------------
+
+def test_patch_done_becomes_revising_when_child_created_via_api(client):
+    """E2E: дочерняя задача создана через POST /api/tasks → parent PATCH done → revising."""
+    r = client.post("/api/tasks", json={"project_id": "p1", "title": "Parent"})
+    parent_id = r.json()["id"]
+
+    client.post("/api/tasks", json={
+        "project_id": "p1",
+        "title": "Child (open)",
+        "parent_task_id": parent_id,
+    })
+
+    r = client.patch(f"/api/tasks/{parent_id}", json={"status": "done"})
+    assert r.status_code == 200
+    assert r.json()["status"] == "revising"
+
+
+def test_patch_done_when_api_child_is_done(client):
+    """E2E: все дочерние задачи (созданные через API) done → parent PATCH done → done."""
+    r = client.post("/api/tasks", json={"project_id": "p1", "title": "Parent"})
+    parent_id = r.json()["id"]
+
+    r = client.post("/api/tasks", json={
+        "project_id": "p1",
+        "title": "Child",
+        "parent_task_id": parent_id,
+    })
+    child_id = r.json()["id"]
+
+    client.patch(f"/api/tasks/{child_id}", json={"status": "done"})
+
+    r = client.patch(f"/api/tasks/{parent_id}", json={"status": "done"})
+    assert r.status_code == 200
+    assert r.json()["status"] == "done"

web/api.py

@@ -660,14 +660,19 @@ def list_tasks(
     limit: int = Query(default=20, ge=1, le=500),
     sort: str = Query(default="updated_at"),
     project_id: str | None = Query(default=None),
+    parent_task_id: str | None = Query(default=None),
 ):
-    """List tasks with optional filters. sort defaults to updated_at desc."""
+    """List tasks with optional filters. sort defaults to updated_at desc.
+
+    parent_task_id: filter by parent task. Use '__none__' to get root tasks only.
+    """
     from core.models import VALID_TASK_SORT_FIELDS
     conn = get_conn()
     tasks = models.list_tasks(
         conn,
         project_id=project_id,
         status=status,
+        parent_task_id=parent_task_id,
         limit=limit,
         sort=sort if sort in VALID_TASK_SORT_FIELDS else "updated_at",
         sort_dir="desc",
@@ -686,6 +691,19 @@ def get_task(task_id: str):
     return t
 
 
+@app.get("/api/tasks/{task_id}/children")
+def get_task_children(task_id: str):
+    """Get direct child tasks of a given task."""
+    conn = get_conn()
+    t = models.get_task(conn, task_id)
+    if not t:
+        conn.close()
+        raise HTTPException(404, f"Task '{task_id}' not found")
+    children = models.get_children(conn, task_id)
+    conn.close()
+    return children
+
+
 class TaskCreate(BaseModel):
     project_id: str
     title: str
@@ -694,6 +712,7 @@ class TaskCreate(BaseModel):
     category: str | None = None
     acceptance_criteria: str | None = None
     labels: list[str] | None = None
+    parent_task_id: str | None = None
 
 
 @app.post("/api/tasks")
@@ -714,7 +733,8 @@ def create_task(body: TaskCreate):
     t = models.create_task(conn, task_id, body.project_id, body.title,
                            priority=body.priority, brief=brief, category=category,
                            acceptance_criteria=body.acceptance_criteria,
-                           labels=body.labels)
+                           labels=body.labels,
+                           parent_task_id=body.parent_task_id)
     conn.close()
 
     # Auto-trigger: if task has 'auto' label, launch pipeline in background