Step 1: implement core Task model and parser/formatter logic

apiad · apiad · commit 449293ce6af2 · 2026-03-24T13:36:58.000-04:00
diff --git a/.gemini/scripts/task.py b/.gemini/scripts/task.py
@@ -0,0 +1,217 @@
+#!/usr/bin/env python3
+
+import re
+import sys
+import argparse
+from collections import defaultdict, deque
+
+class Task:
+    def __init__(self, id=None, label=None, description=None, category=None, complexity=0, dependencies=None, status="todo", plan_path=None):
+        self.id = id
+        self.label = label
+        self.description = description
+        self.category = category
+        self.complexity = complexity
+        self.dependencies = dependencies if dependencies is not None else []
+        self.status = status # e.g., "todo", "in_progress", "done"
+        self.plan_path = plan_path
+
+    def __repr__(self):
+        return (f"Task(id={self.id!r}, label={self.label!r}, description={self.description!r}, category={self.category!r}, "
+                f"complexity={self.complexity!r}, dependencies={self.dependencies!r}, status={self.status!r}, plan_path={self.plan_path!r})")
+
+    def __eq__(self, other):
+        if not isinstance(other, Task):
+            return NotImplemented
+        return (self.id == other.id and
+                self.label == other.label and
+                self.description == other.description and
+                self.category == other.category and
+                self.complexity == other.complexity and
+                self.dependencies == other.dependencies and
+                self.status == other.status and
+                self.plan_path == other.plan_path)
+
+# --- Regex Patterns ---
+# New format: - [Status] **[ID]** Label: Description (Complexity: X) [Deps: Y] (See plan: Z)
+NEW_TASK_REGEX = re.compile(
+    r'^- \[(?P<status>[^\]]+)\] \*\*(?P<id>[^ ]+)\*\* (?P<label>[^:]+): (?P<description>.*?)(?: \(Complexity: (?P<complexity>\d+)\))?(?: \[Deps: (?P<dependencies>.*?)\])?(?: \(See plan: (?P<plan_path>.*?)\))?$'
+)
+# Old format: - [ ] Description (See plan: ...)
+OLD_TASK_REGEX = re.compile(
+    r'^- \[(?P<status>[^\]]*)\] (?P<description>.*?)(?: \(See plan: (?P<plan_path>.*?)\))?$'
+)
+HEADER_WARNING = """# Tasks
+
+> **WARNING: NEVER MODIFY THIS FILE BY HAND. USE THE SCRIPT INSTEAD.**
+> Run `python .gemini/scripts/task.py --help` for usage.
+"""
+
+# --- Parser Functions ---
+
+def parse_task_line(line):
+    line = line.strip()
+    if not line or line.startswith('#') or line.startswith('>'):
+        return None
+
+    new_match = NEW_TASK_REGEX.match(line)
+    if new_match:
+        data = new_match.groupdict()
+        deps_str = data.get('dependencies')
+        dependencies = []
+        if deps_str:
+            dependencies = [dep.strip() for dep in deps_str.split(',') if dep.strip()]
+        
+        complexity = int(data.get('complexity')) if data.get('complexity') else 0
+        return Task(
+            id=data.get('id'),
+            label=data.get('label'),
+            description=data.get('description'),
+            category=None,
+            complexity=complexity,
+            dependencies=dependencies,
+            status=data.get('status'),
+            plan_path=data.get('plan_path')
+        )
+
+    old_match = OLD_TASK_REGEX.match(line)
+    if old_match:
+        data = old_match.groupdict()
+        status = data.get('status').strip()
+        if not status: status = "todo"
+        return Task(
+            id=None,
+            label=None,
+            description=data.get('description'),
+            category=None,
+            complexity=0,
+            dependencies=[],
+            status=status,
+            plan_path=data.get('plan_path')
+        )
+    return None
+
+def parse_tasks_file(file_content):
+    tasks = []
+    current_category = None
+    lines = file_content.splitlines()
+
+    for line in lines:
+        stripped_line = line.strip()
+        if stripped_line.startswith("## Active Tasks"):
+            current_category = None
+            continue
+        elif stripped_line.startswith("## Archive"):
+            current_category = None
+            continue
+        elif stripped_line.startswith("### "):
+            current_category = stripped_line[4:].strip()
+            continue
+        
+        task = parse_task_line(line)
+        if task:
+            task.category = current_category
+            tasks.append(task)
+    return tasks
+
+# --- Formatter Functions ---
+
+def format_task_to_line(task):
+    if task.id is not None:
+        complexity_str = f" (Complexity: {task.complexity})" if task.complexity != 0 else ""
+        deps_str = f" [Deps: {', '.join(task.dependencies)}]"
+        plan_path_str = f" (See plan: {task.plan_path})" if task.plan_path else ""
+        return f"- [{task.status}] **{task.id}** {task.label}: {task.description}{complexity_str}{deps_str}{plan_path_str}"
+    else:
+        plan_path_str = f" (See plan: {task.plan_path})" if task.plan_path else ""
+        return f"- [{task.status}] {task.description}{plan_path_str}"
+
+def topological_sort(tasks):
+    adj = defaultdict(list)
+    in_degree = defaultdict(int)
+    task_map = {task.id: task for task in tasks if task.id}
+
+    for tid in task_map:
+        in_degree[tid] = 0
+
+    for task in tasks:
+        if task.id:
+            for dep_id in task.dependencies:
+                if dep_id in task_map:
+                    adj[dep_id].append(task.id)
+                    in_degree[task.id] += 1
+
+    queue = deque(sorted([tid for tid in in_degree if in_degree[tid] == 0]))
+    sorted_tasks_ids = []
+
+    while queue:
+        u = queue.popleft()
+        sorted_tasks_ids.append(u)
+        for v in sorted(adj[u]):
+            in_degree[v] -= 1
+            if in_degree[v] == 0:
+                queue.append(v)
+
+    sorted_tasks = [task_map[tid] for tid in sorted_tasks_ids]
+    # Add tasks with IDs that were not in the sort (cycles)
+    remaining_with_ids = sorted([t for t in tasks if t.id and t.id not in sorted_tasks_ids], key=lambda t: (t.complexity, t.id))
+    # Add tasks without IDs
+    without_ids = sorted([t for t in tasks if not t.id], key=lambda t: (t.complexity, t.description))
+    
+    return sorted_tasks + remaining_with_ids + without_ids
+
+def format_tasks_to_markdown(tasks):
+    active_tasks = [t for t in tasks if t.status != "done"]
+    archive_tasks = [t for t in tasks if t.status == "done"]
+
+    def group_by_category(task_list):
+        grouped = defaultdict(list)
+        for t in task_list:
+            cat = t.category if t.category else "Uncategorized"
+            grouped[cat].append(t)
+        return grouped
+
+    active_grouped = group_by_category(active_tasks)
+    archive_grouped = group_by_category(archive_tasks)
+
+    lines = [HEADER_WARNING.strip(), ""]
+    
+    lines.append("## Active Tasks")
+    if not active_tasks:
+        lines.append("No active tasks.")
+    else:
+        for cat in sorted(active_grouped.keys()):
+            lines.append(f"### {cat}")
+            for t in topological_sort(active_grouped[cat]):
+                lines.append(format_task_to_line(t))
+    
+    lines.append("## Archive")
+    if not archive_tasks:
+        lines.append("No archived tasks.")
+    else:
+        for cat in sorted(archive_grouped.keys()):
+            lines.append(f"### {cat}")
+            # Archive sorted by complexity then id/description
+            cat_tasks = sorted(archive_grouped[cat], key=lambda t: (t.complexity, t.id if t.id else t.description))
+            for t in cat_tasks:
+                lines.append(format_task_to_line(t))
+    
+    return "\n".join(lines) + "\n"
+
+def main():
+    parser = argparse.ArgumentParser(description="Task management script for TASKS.md.")
+    # For now we just implement the reformat on call
+    tasks_file_path = "TASKS.md"
+    try:
+        with open(tasks_file_path, 'r') as f:
+            content = f.read()
+    except FileNotFoundError:
+        sys.exit(1)
+
+    tasks = parse_tasks_file(content)
+    formatted = format_tasks_to_markdown(tasks)
+    with open(tasks_file_path, 'w') as f:
+        f.write(formatted)
+
+if __name__ == "__main__":
+    main()
diff --git a/journal/2026-03-24.md b/journal/2026-03-24.md
@@ -3,3 +3,4 @@
 [2026-03-24T13:16:16] - Added Tier Protocol RCA for 2026-03-24
 [2026-03-24T13:27:32] - Starting implementation of procedural task management
 [2026-03-24T13:27:52] - Started procedural task management implementation
+[2026-03-24T13:36:58] - Implemented core Task model and parser/formatter logic
diff --git a/tests/test_task_script.py b/tests/test_task_script.py
@@ -0,0 +1,168 @@
+import unittest
+import sys
+import os
+
+# Add the script directory to the path so we can import from it
+sys.path.append(os.path.join(os.path.dirname(__file__), '..', '.gemini', 'scripts'))
+from task import Task, parse_task_line, parse_tasks_file, format_task_to_line, format_tasks_to_markdown, topological_sort
+
+class TestTaskScript(unittest.TestCase):
+
+    # --- Parser Tests ---
+
+    def test_parse_new_task_format(self):
+        line = '- [todo] **B.1** Setup Database: Initialize the PostgreSQL schema (Complexity: 3) [Deps: ] (See plan: plans/db.md)'
+        task = parse_task_line(line)
+        self.assertIsNotNone(task)
+        self.assertEqual(task.id, 'B.1')
+        self.assertEqual(task.label, 'Setup Database')
+        self.assertEqual(task.description, 'Initialize the PostgreSQL schema')
+        self.assertEqual(task.complexity, 3)
+        self.assertEqual(task.dependencies, [])
+        self.assertEqual(task.status, 'todo')
+        self.assertEqual(task.plan_path, 'plans/db.md')
+
+    def test_parse_new_task_format_multiple_deps(self):
+        line = '- [in_progress] **U.2** UI Components: Create shared buttons (Complexity: 1) [Deps: U.1, G.5]'
+        task = parse_task_line(line)
+        self.assertIsNotNone(task)
+        self.assertEqual(task.dependencies, ['U.1', 'G.5'])
+        self.assertEqual(task.status, 'in_progress')
+
+    def test_parse_old_task_format(self):
+        line = '- [ ] Implement auth system (See plan: plans/auth.md)'
+        task = parse_task_line(line)
+        self.assertIsNotNone(task)
+        self.assertIsNone(task.id)
+        self.assertEqual(task.description, 'Implement auth system')
+        self.assertEqual(task.status, 'todo')
+        self.assertEqual(task.plan_path, 'plans/auth.md')
+
+    def test_parse_tasks_file_structure(self):
+        content = """# Tasks
+
+> WARNING: NEVER MODIFY BY HAND
+
+## Active Tasks
+### Backend
+- [todo] **B.1** Task 1: Desc 1
+- [todo] **B.2** Task 2: Desc 2
+
+### Frontend
+- [in_progress] **F.1** Task 3: Desc 3
+
+## Archive
+### Backend
+- [done] **B.3** Task 4: Desc 4
+"""
+        tasks = parse_tasks_file(content)
+        self.assertEqual(len(tasks), 4)
+        self.assertEqual(tasks[0].category, 'Backend')
+        self.assertEqual(tasks[0].id, 'B.1')
+        self.assertEqual(tasks[2].category, 'Frontend')
+        self.assertEqual(tasks[3].category, 'Backend')
+        self.assertEqual(tasks[3].status, 'done')
+
+    # --- Formatting Tests ---
+
+    def test_format_task_to_markdown_line(self):
+        task = Task(
+            id="T1",
+            label="Task 1",
+            description="Desc 1",
+            category="Backend",
+            complexity=1,
+            dependencies=[],
+            status="todo",
+            plan_path="plans/t1.md"
+        )
+        expected_line = '- [todo] **T1** Task 1: Desc 1 (Complexity: 1) [Deps: ] (See plan: plans/t1.md)'
+        formatted_line = format_task_to_line(task)
+        self.assertEqual(formatted_line, expected_line)
+
+    def test_format_task_to_markdown_line_with_zero_complexity(self):
+        task = Task(
+            id="T1",
+            label="Task 1",
+            description="Desc 1",
+            category="Backend",
+            complexity=0,
+            dependencies=[],
+            status="todo",
+            plan_path="plans/t1.md"
+        )
+        expected_line = '- [todo] **T1** Task 1: Desc 1 [Deps: ] (See plan: plans/t1.md)'
+        formatted_line = format_task_to_line(task)
+        self.assertEqual(formatted_line, expected_line)
+
+    def test_format_tasks_to_markdown_with_sorting_and_grouping(self):
+        tasks = [
+            Task(id="T3", label="Task C", description="Desc C", category="Frontend", complexity=1, status="done"),
+            Task(id="T1", label="Task A", description="Desc A", category="Backend", complexity=2, dependencies=["T2"], status="todo"),
+            Task(id="T2", label="Task B", description="Desc B", category="Backend", complexity=1, dependencies=[], status="todo"),
+            Task(id="T4", label="Task D", description="Desc D", category="Frontend", complexity=2, status="in_progress"),
+            Task(id="T5", label="Task E", description="Desc E", category="Backend", complexity=1, dependencies=["T1"], status="todo"),
+        ]
+        
+        expected_markdown = """# Tasks
+
+> **WARNING: NEVER MODIFY THIS FILE BY HAND. USE THE SCRIPT INSTEAD.**
+> Run `python .gemini/scripts/task.py --help` for usage.
+
+## Active Tasks
+### Backend
+- [todo] **T2** Task B: Desc B (Complexity: 1) [Deps: ]
+- [todo] **T1** Task A: Desc A (Complexity: 2) [Deps: T2]
+- [todo] **T5** Task E: Desc E (Complexity: 1) [Deps: T1]
+### Frontend
+- [in_progress] **T4** Task D: Desc D (Complexity: 2) [Deps: ]
+## Archive
+### Frontend
+- [done] **T3** Task C: Desc C (Complexity: 1) [Deps: ]
+"""
+        markdown_output = format_tasks_to_markdown(tasks)
+        self.assertEqual(markdown_output, expected_markdown)
+
+    def test_format_tasks_to_markdown_with_unsorted_categories_and_mixed_statuses(self):
+        tasks = [
+            Task(id="T1", label="Task Alpha", description="Desc Alpha", category="Zebra", complexity=1, status="todo"),
+            Task(id="T2", label="Task Beta", description="Desc Beta", category="Apple", complexity=2, dependencies=["T1"], status="in_progress"),
+            Task(id="T3", label="Task Gamma", description="Desc Gamma", category="Zebra", complexity=3, status="done"),
+            Task(id="T4", label="Task Delta", description="Desc Delta", category="Apple", complexity=1, status="todo"),
+        ]
+        
+        expected_markdown = """# Tasks
+
+> **WARNING: NEVER MODIFY THIS FILE BY HAND. USE THE SCRIPT INSTEAD.**
+> Run `python .gemini/scripts/task.py --help` for usage.
+
+## Active Tasks
+### Apple
+- [in_progress] **T2** Task Beta: Desc Beta (Complexity: 2) [Deps: T1]
+- [todo] **T4** Task Delta: Desc Delta (Complexity: 1) [Deps: ]
+### Zebra
+- [todo] **T1** Task Alpha: Desc Alpha (Complexity: 1) [Deps: ]
+## Archive
+### Zebra
+- [done] **T3** Task Gamma: Desc Gamma (Complexity: 3) [Deps: ]
+"""
+        markdown_output = format_tasks_to_markdown(tasks)
+        self.assertEqual(markdown_output, expected_markdown)
+
+    def test_format_tasks_to_markdown_with_no_tasks(self):
+        tasks = []
+        expected_markdown = """# Tasks
+
+> **WARNING: NEVER MODIFY THIS FILE BY HAND. USE THE SCRIPT INSTEAD.**
+> Run `python .gemini/scripts/task.py --help` for usage.
+
+## Active Tasks
+No active tasks.
+## Archive
+No archived tasks.
+"""
+        markdown_output = format_tasks_to_markdown(tasks)
+        self.assertEqual(markdown_output, expected_markdown)
+
+if __name__ == '__main__':
+    unittest.main()
