feat(web-ui): task → PRD requirement traceability badges (#468)

Closes #468

Adds requirement_ids linkage between tasks and PROOF9 requirements for full traceability.

- Task dataclass: requirement_ids field, persisted as JSON in SQLite with migration guard
- API: TaskResponse model updated, update_requirement_ids() function added
- Frontend: useRequirementsLookup SWR hook (memoized Map), TaskCard badges, TaskDetailModal section
- Tests: 8 new v2 tests including real migration simulation

Author: frankbria

codeframe/core/tasks.py

@@ -60,6 +60,7 @@ class Task:
     lineage: list[str] = field(default_factory=list)
     is_leaf: bool = True
     hierarchical_id: Optional[str] = None
+    requirement_ids: list[str] = field(default_factory=list)
 
 
 def create(
@@ -77,6 +78,7 @@ def create(
     lineage: Optional[list[str]] = None,
     is_leaf: bool = True,
     hierarchical_id: Optional[str] = None,
+    requirement_ids: Optional[list[str]] = None,
 ) -> Task:
     """Create a new task.
 
@@ -95,6 +97,7 @@ def create(
         lineage: Optional list of ancestor descriptions
         is_leaf: Whether this is a leaf/executable task (default True)
         hierarchical_id: Optional display ID like "1.2.3"
+        requirement_ids: Optional list of PROOF9 requirement IDs this task implements
 
     Returns:
         Created Task
@@ -103,16 +106,17 @@ def create(
     now = _utc_now().isoformat()
     depends_on_list = depends_on or []
     lineage_list = lineage or []
+    requirement_ids_list = requirement_ids or []
 
     conn = get_db_connection(workspace)
     try:
         cursor = conn.cursor()
         cursor.execute(
             """
-            INSERT INTO tasks (id, workspace_id, prd_id, title, description, status, priority, depends_on, estimated_hours, complexity_score, uncertainty_level, parent_id, lineage, is_leaf, hierarchical_id, created_at, updated_at)
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            INSERT INTO tasks (id, workspace_id, prd_id, title, description, status, priority, depends_on, estimated_hours, complexity_score, uncertainty_level, parent_id, lineage, is_leaf, hierarchical_id, created_at, updated_at, requirement_ids)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
             """,
-            (task_id, workspace.id, prd_id, title, description, status.value, priority, json.dumps(depends_on_list), estimated_hours, complexity_score, uncertainty_level, parent_id, json.dumps(lineage_list), 1 if is_leaf else 0, hierarchical_id, now, now),
+            (task_id, workspace.id, prd_id, title, description, status.value, priority, json.dumps(depends_on_list), estimated_hours, complexity_score, uncertainty_level, parent_id, json.dumps(lineage_list), 1 if is_leaf else 0, hierarchical_id, now, now, json.dumps(requirement_ids_list)),
         )
         conn.commit()
     finally:
@@ -134,6 +138,7 @@ def create(
         lineage=lineage_list,
         is_leaf=is_leaf,
         hierarchical_id=hierarchical_id,
+        requirement_ids=requirement_ids_list,
         created_at=datetime.fromisoformat(now),
         updated_at=datetime.fromisoformat(now),
     )
@@ -154,7 +159,7 @@ def get(workspace: Workspace, task_id: str) -> Optional[Task]:
 
     cursor.execute(
         """
-        SELECT id, workspace_id, prd_id, title, description, status, priority, depends_on, estimated_hours, complexity_score, uncertainty_level, created_at, updated_at, github_issue_number, parent_id, lineage, is_leaf, hierarchical_id
+        SELECT id, workspace_id, prd_id, title, description, status, priority, depends_on, estimated_hours, complexity_score, uncertainty_level, created_at, updated_at, github_issue_number, parent_id, lineage, is_leaf, hierarchical_id, requirement_ids
         FROM tasks
         WHERE workspace_id = ? AND id = ?
         """,
@@ -190,7 +195,7 @@ def list_tasks(
     if status:
         cursor.execute(
             """
-            SELECT id, workspace_id, prd_id, title, description, status, priority, depends_on, estimated_hours, complexity_score, uncertainty_level, created_at, updated_at, github_issue_number, parent_id, lineage, is_leaf, hierarchical_id
+            SELECT id, workspace_id, prd_id, title, description, status, priority, depends_on, estimated_hours, complexity_score, uncertainty_level, created_at, updated_at, github_issue_number, parent_id, lineage, is_leaf, hierarchical_id, requirement_ids
             FROM tasks
             WHERE workspace_id = ? AND status = ?
             ORDER BY priority ASC, created_at ASC
@@ -201,7 +206,7 @@ def list_tasks(
     else:
         cursor.execute(
             """
-            SELECT id, workspace_id, prd_id, title, description, status, priority, depends_on, estimated_hours, complexity_score, uncertainty_level, created_at, updated_at, github_issue_number, parent_id, lineage, is_leaf, hierarchical_id
+            SELECT id, workspace_id, prd_id, title, description, status, priority, depends_on, estimated_hours, complexity_score, uncertainty_level, created_at, updated_at, github_issue_number, parent_id, lineage, is_leaf, hierarchical_id, requirement_ids
             FROM tasks
             WHERE workspace_id = ?
             ORDER BY priority ASC, created_at ASC
@@ -415,6 +420,49 @@ def update_depends_on(
     return task
 
 
+def update_requirement_ids(
+    workspace: Workspace,
+    task_id: str,
+    requirement_ids: list[str],
+) -> Task:
+    """Update a task's linked PROOF9 requirement IDs.
+
+    Args:
+        workspace: Target workspace
+        task_id: Task to update
+        requirement_ids: List of PROOF9 requirement IDs this task implements
+
+    Returns:
+        Updated Task
+
+    Raises:
+        ValueError: If task not found
+    """
+    task = get(workspace, task_id)
+    if not task:
+        raise ValueError(f"Task not found: {task_id}")
+
+    now = _utc_now().isoformat()
+
+    conn = get_db_connection(workspace)
+    cursor = conn.cursor()
+    cursor.execute(
+        """
+        UPDATE tasks
+        SET requirement_ids = ?, updated_at = ?
+        WHERE workspace_id = ? AND id = ?
+        """,
+        (json.dumps(requirement_ids), now, workspace.id, task_id),
+    )
+    conn.commit()
+    conn.close()
+
+    task.requirement_ids = requirement_ids
+    task.updated_at = datetime.fromisoformat(now)
+
+    return task
+
+
 def get_dependents(workspace: Workspace, task_id: str) -> list[Task]:
     """Get all tasks that depend on the given task.
 
@@ -705,7 +753,7 @@ def _row_to_task(row: tuple) -> Task:
     Row columns: id, workspace_id, prd_id, title, description, status, priority,
                  depends_on, estimated_hours, complexity_score, uncertainty_level,
                  created_at, updated_at, github_issue_number, parent_id, lineage,
-                 is_leaf, hierarchical_id
+                 is_leaf, hierarchical_id, requirement_ids
     """
     # Parse depends_on from JSON string (default to empty list if null)
     depends_on_raw = row[7]
@@ -719,6 +767,10 @@ def _row_to_task(row: tuple) -> Task:
     is_leaf_raw = row[16] if len(row) > 16 else 1
     is_leaf = bool(is_leaf_raw) if is_leaf_raw is not None else True
 
+    # Parse requirement_ids from JSON string (default to empty list if null)
+    requirement_ids_raw = row[18] if len(row) > 18 else None
+    requirement_ids = json.loads(requirement_ids_raw) if requirement_ids_raw else []
+
     return Task(
         id=row[0],
         workspace_id=row[1],
@@ -738,4 +790,5 @@ def _row_to_task(row: tuple) -> Task:
         lineage=lineage,
         is_leaf=is_leaf,
         hierarchical_id=row[17] if len(row) > 17 else None,
+        requirement_ids=requirement_ids,
     )

codeframe/core/workspace.py

@@ -147,6 +147,8 @@ def _init_database(db_path: Path) -> None:
         cursor.execute("ALTER TABLE tasks ADD COLUMN is_leaf INTEGER DEFAULT 1")
     if "hierarchical_id" not in columns:
         cursor.execute("ALTER TABLE tasks ADD COLUMN hierarchical_id TEXT")
+    if "requirement_ids" not in columns:
+        cursor.execute("ALTER TABLE tasks ADD COLUMN requirement_ids TEXT DEFAULT '[]'")
 
     # Append-only event log
     cursor.execute("""
@@ -489,6 +491,9 @@ def _ensure_schema_upgrades(db_path: Path) -> None:
         if "hierarchical_id" not in task_columns:
             cursor.execute("ALTER TABLE tasks ADD COLUMN hierarchical_id TEXT")
             conn.commit()
+        if "requirement_ids" not in task_columns:
+            cursor.execute("ALTER TABLE tasks ADD COLUMN requirement_ids TEXT DEFAULT '[]'")
+            conn.commit()
         if "github_issue_number" not in task_columns:
             cursor.execute("ALTER TABLE tasks ADD COLUMN github_issue_number INTEGER")
             conn.commit()

codeframe/ui/routers/tasks_v2.py

@@ -141,6 +141,7 @@ class TaskResponse(BaseModel):
     status: str
     priority: int
     depends_on: list[str] = []
+    requirement_ids: list[str] = []
     estimated_hours: Optional[float] = None
     created_at: Optional[str] = None
     updated_at: Optional[str] = None
@@ -216,6 +217,7 @@ async def list_tasks(
                 status=t.status.value,
                 priority=t.priority,
                 depends_on=t.depends_on,
+                requirement_ids=t.requirement_ids,
                 estimated_hours=t.estimated_hours,
                 created_at=t.created_at.isoformat() if t.created_at else None,
                 updated_at=t.updated_at.isoformat() if t.updated_at else None,
@@ -260,6 +262,7 @@ async def get_task(
         status=task.status.value,
         priority=task.priority,
         depends_on=task.depends_on,
+        requirement_ids=task.requirement_ids,
         estimated_hours=task.estimated_hours,
         created_at=task.created_at.isoformat() if task.created_at else None,
         updated_at=task.updated_at.isoformat() if task.updated_at else None,
@@ -330,6 +333,7 @@ async def update_task(
             status=task.status.value,
             priority=task.priority,
             depends_on=task.depends_on,
+            requirement_ids=task.requirement_ids,
             estimated_hours=task.estimated_hours,
             created_at=task.created_at.isoformat() if task.created_at else None,
             updated_at=task.updated_at.isoformat() if task.updated_at else None,

tests/core/test_task_requirement_ids.py

@@ -0,0 +1,139 @@
+"""Tests for task requirement_ids field (issue #468).
+
+Tests that tasks can be linked to PROOF9 requirement IDs for traceability.
+"""
+
+import sqlite3
+
+import pytest
+
+from codeframe.core import tasks
+from codeframe.core.workspace import create_or_load_workspace
+
+pytestmark = pytest.mark.v2
+
+
+@pytest.fixture
+def workspace(tmp_path):
+    """Create a test workspace."""
+    return create_or_load_workspace(tmp_path)
+
+
+class TestTaskRequirementIdsField:
+    """Test the requirement_ids field on Task model."""
+
+    def test_task_has_empty_requirement_ids_by_default(self, workspace):
+        """New tasks should have empty requirement_ids list."""
+        task = tasks.create(workspace, title="Test task")
+        assert task.requirement_ids == []
+
+    def test_task_created_with_requirement_ids(self, workspace):
+        """Tasks can be created with requirement_ids."""
+        req_ids = ["REQ-001", "REQ-002"]
+        task = tasks.create(workspace, title="Task with reqs", requirement_ids=req_ids)
+        assert task.requirement_ids == req_ids
+
+    def test_task_get_includes_requirement_ids(self, workspace):
+        """Getting a task should include its requirement_ids."""
+        req_ids = ["REQ-042"]
+        task = tasks.create(workspace, title="Task", requirement_ids=req_ids)
+        retrieved = tasks.get(workspace, task.id)
+        assert retrieved.requirement_ids == req_ids
+
+    def test_task_list_includes_requirement_ids(self, workspace):
+        """Listing tasks should include requirement_ids."""
+        t1 = tasks.create(workspace, title="No reqs")
+        t2 = tasks.create(workspace, title="With reqs", requirement_ids=["REQ-007"])
+
+        all_tasks = tasks.list_tasks(workspace)
+        task_map = {t.id: t for t in all_tasks}
+
+        assert task_map[t1.id].requirement_ids == []
+        assert task_map[t2.id].requirement_ids == ["REQ-007"]
+
+    def test_task_requirement_ids_persisted_across_get(self, workspace):
+        """requirement_ids should survive a round-trip to the database."""
+        req_ids = ["REQ-001", "REQ-002", "REQ-003"]
+        task = tasks.create(workspace, title="Multi-req task", requirement_ids=req_ids)
+        fetched = tasks.get(workspace, task.id)
+        assert fetched.requirement_ids == req_ids
+
+    def test_update_requirement_ids(self, workspace):
+        """requirement_ids can be updated on an existing task."""
+        task = tasks.create(workspace, title="Task")
+        assert task.requirement_ids == []
+
+        updated = tasks.update_requirement_ids(workspace, task.id, ["REQ-099"])
+        assert updated.requirement_ids == ["REQ-099"]
+
+        fetched = tasks.get(workspace, task.id)
+        assert fetched.requirement_ids == ["REQ-099"]
+
+    def test_update_requirement_ids_to_empty(self, workspace):
+        """requirement_ids can be cleared."""
+        task = tasks.create(workspace, title="Task", requirement_ids=["REQ-001"])
+        updated = tasks.update_requirement_ids(workspace, task.id, [])
+        assert updated.requirement_ids == []
+
+    def test_task_without_requirement_ids_in_existing_db(self, tmp_path):
+        """Migration guard adds requirement_ids column to pre-migration databases.
+
+        Simulates a workspace that was created before the requirement_ids column
+        was added, then verifies that opening it triggers the migration guard
+        and tasks can be read back with requirement_ids == [].
+        """
+        # Step 1: Create a workspace and inject a "pre-migration" tasks table
+        # by directly dropping the requirement_ids column from the DB.
+        ws = create_or_load_workspace(tmp_path)
+        db_path = ws.db_path
+
+        conn = sqlite3.connect(db_path)
+        # Insert a task using the old schema (without requirement_ids)
+        import uuid
+        from datetime import datetime, timezone
+        task_id = str(uuid.uuid4())
+        now = datetime.now(timezone.utc).isoformat()
+        conn.execute(
+            """
+            INSERT INTO tasks (id, workspace_id, prd_id, title, description, status,
+                               priority, depends_on, created_at, updated_at)
+            VALUES (?, ?, NULL, ?, '', 'BACKLOG', 0, '[]', ?, ?)
+            """,
+            (task_id, ws.id, "Legacy task", now, now),
+        )
+        # Simulate the column not existing by removing it (SQLite workaround)
+        conn.execute("ALTER TABLE tasks RENAME TO tasks_old")
+        conn.execute("""
+            CREATE TABLE tasks (
+                id TEXT PRIMARY KEY,
+                workspace_id TEXT NOT NULL,
+                prd_id TEXT,
+                title TEXT NOT NULL,
+                description TEXT,
+                status TEXT NOT NULL DEFAULT 'BACKLOG',
+                priority INTEGER DEFAULT 0,
+                depends_on TEXT DEFAULT '[]',
+                created_at TEXT NOT NULL,
+                updated_at TEXT NOT NULL
+            )
+        """)
+        conn.execute("""
+            INSERT INTO tasks (id, workspace_id, prd_id, title, description, status,
+                               priority, depends_on, created_at, updated_at)
+            SELECT id, workspace_id, prd_id, title, description, status,
+                   priority, depends_on, created_at, updated_at
+            FROM tasks_old
+        """)
+        conn.execute("DROP TABLE tasks_old")
+        conn.commit()
+        conn.close()
+
+        # Step 2: Re-open workspace — this triggers _ensure_schema_upgrades()
+        # which should add the requirement_ids column.
+        ws2 = create_or_load_workspace(tmp_path)
+
+        # Step 3: Verify the legacy task reads back with requirement_ids == []
+        fetched = tasks.get(ws2, task_id)
+        assert fetched is not None
+        assert hasattr(fetched, "requirement_ids")
+        assert fetched.requirement_ids == []

web-ui/src/components/tasks/TaskBoardContent.tsx

@@ -2,7 +2,7 @@
 
 import { useMemo } from 'react';
 import { TaskColumn } from './TaskColumn';
-import type { Task, TaskStatus } from '@/types';
+import type { Task, TaskStatus, ProofRequirement } from '@/types';
 
 /** Column display order matches the task lifecycle. */
 const COLUMN_ORDER: TaskStatus[] = [
@@ -27,6 +27,7 @@ interface TaskBoardContentProps {
   onSelectAll?: (taskIds: string[]) => void;
   onDeselectAll?: (taskIds: string[]) => void;
   loadingTaskIds?: Set<string>;
+  requirementsMap?: Map<string, ProofRequirement>;
 }
 
 export function TaskBoardContent({
@@ -42,6 +43,7 @@ export function TaskBoardContent({
   onSelectAll,
   onDeselectAll,
   loadingTaskIds,
+  requirementsMap,
 }: TaskBoardContentProps) {
   /** Group flat task array into per-status buckets. */
   const tasksByStatus = useMemo(() => {
@@ -75,6 +77,7 @@ export function TaskBoardContent({
           onSelectAll={onSelectAll}
           onDeselectAll={onDeselectAll}
           loadingTaskIds={loadingTaskIds}
+          requirementsMap={requirementsMap}
         />
       ))}
     </div>