feature-pipeline — Skillopedia

Feature Pipeline Execute implementation tasks directly from design documents. Tasks are managed as markdown checkboxes - no separate session files needed. Quick Reference Task Format Tasks are written as markdown checkboxes in the design document: See references/task-format.md for full format specification. Execution Loop Unattended Mode Rules - NO stopping for questions - NO asking for clarification - Make autonomous decisions based on codebase patterns - If blocked, mark as failed and continue Status Updates Completed task: Failed task: Resume / Recovery To resume interrupted work, simply r…

, line.strip())\n if not match:\n return None\n\n checkbox, title, rest = match.groups()\n status = \"completed\" if checkbox.lower() == 'x' else \"pending\"\n\n # Check for failure marker\n if \"❌\" in rest or \"FAILED\" in rest:\n status = \"failed\"\n\n # Parse inline attributes\n priority = 5\n phase = \"implementation\"\n dependencies = []\n\n # Extract priority\n priority_match = re.search(r'`priority:(\\d+)`', rest)\n if priority_match:\n priority = int(priority_match.group(1))\n\n # Extract phase\n phase_match = re.search(r'`phase:(\\w+)`', rest)\n if phase_match:\n phase = phase_match.group(1)\n\n # Extract dependencies\n deps_match = re.search(r'`deps:([^`]+)`', rest)\n if deps_match:\n dependencies = [d.strip() for d in deps_match.group(1).split(',')]\n\n return {\n \"title\": title.strip(),\n \"status\": status,\n \"priority\": priority,\n \"phase\": phase,\n \"dependencies\": dependencies\n }\n\n\ndef parse_tasks_from_markdown(content: str) -> list[Task]:\n \"\"\"Parse all tasks from markdown content.\"\"\"\n\n lines = content.split('\\n')\n tasks = []\n current_task = None\n in_task_section = False\n\n for i, line in enumerate(lines):\n # Check if we're in the Implementation Tasks section\n if re.match(r'^##\\s+Implementation\\s+Tasks', line, re.IGNORECASE):\n in_task_section = True\n continue\n\n # Exit task section on next ## header\n if in_task_section and re.match(r'^##\\s+[^#]', line) and 'Implementation' not in line:\n in_task_section = False\n continue\n\n if not in_task_section:\n continue\n\n # Parse main task line\n task_data = parse_task_line(line)\n if task_data:\n if current_task:\n tasks.append(current_task)\n\n current_task = Task(\n title=task_data[\"title\"],\n status=task_data[\"status\"],\n priority=task_data[\"priority\"],\n phase=task_data[\"phase\"],\n dependencies=task_data[\"dependencies\"],\n line_number=i + 1\n )\n continue\n\n # Parse task details (indented lines under a task)\n if current_task and line.strip().startswith('- '):\n stripped = line.strip()\n\n # Files line\n if stripped.startswith('- files:'):\n files_str = stripped.replace('- files:', '').strip()\n current_task.files = [f.strip() for f in files_str.split(',') if f.strip()]\n\n # Criterion line (checkbox)\n elif re.match(r'^- \\[([ xX])\\] ', stripped):\n checkbox_match = re.match(r'^- \\[([ xX])\\] (.+)

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.

, stripped)\n if checkbox_match:\n is_done = checkbox_match.group(1).lower() == 'x'\n criterion = checkbox_match.group(2).strip()\n current_task.criteria.append(criterion)\n current_task.criteria_status.append(is_done)\n\n # Failure reason\n elif stripped.startswith('- reason:') or stripped.startswith('- error:'):\n current_task.failure_reason = stripped.split(':', 1)[1].strip()\n\n # Don't forget the last task\n if current_task:\n tasks.append(current_task)\n\n return tasks\n\n\ndef get_next_task(tasks: list[Task]) -> Optional[Task]:\n \"\"\"Get the next task to execute based on priority and dependencies.\"\"\"\n\n completed_titles = {t.title for t in tasks if t.status == \"completed\"}\n\n # Find pending tasks with satisfied dependencies\n available = []\n for task in tasks:\n if task.status != \"pending\":\n continue\n\n # Check dependencies\n deps_satisfied = all(dep in completed_titles for dep in task.dependencies)\n if deps_satisfied:\n available.append(task)\n\n if not available:\n return None\n\n # Sort by priority (lower number = higher priority)\n available.sort(key=lambda t: t.priority)\n return available[0]\n\n\ndef update_task_status(content: str, task_title: str, new_status: str, reason: str = \"\") -> str:\n \"\"\"Update a task's status in the markdown content.\"\"\"\n\n lines = content.split('\\n')\n result = []\n in_target_task = False\n task_indent = 0\n\n for line in lines:\n # Check if this is the target task\n task_data = parse_task_line(line)\n if task_data and task_data[\"title\"] == task_title:\n in_target_task = True\n task_indent = len(line) - len(line.lstrip())\n\n # Update the checkbox\n if new_status == \"completed\":\n line = re.sub(r'^(\\s*- )\\[[ ]\\]', r'\\1[x]', line)\n # Add completion marker if not present\n if \"✅\" not in line:\n line = line.rstrip() + \" ✅\"\n elif new_status == \"failed\":\n line = re.sub(r'^(\\s*- )\\[[ ]\\]', r'\\1[x]', line)\n # Add failure marker\n if \"❌\" not in line:\n line = line.rstrip() + \" ❌\"\n elif new_status == \"pending\":\n line = re.sub(r'^(\\s*- )\\[[xX]\\]', r'\\1[ ]', line)\n # Remove markers\n line = line.replace(\" ✅\", \"\").replace(\" ❌\", \"\")\n\n result.append(line)\n continue\n\n # Check if we've moved to a different task\n if task_data and task_data[\"title\"] != task_title:\n in_target_task = False\n\n # Update criteria checkboxes within the task\n if in_target_task and re.match(r'^\\s*- \\[[ xX]\\] ', line):\n current_indent = len(line) - len(line.lstrip())\n if current_indent > task_indent:\n if new_status == \"completed\":\n line = re.sub(r'^(\\s*- )\\[[ ]\\]', r'\\1[x]', line)\n elif new_status == \"pending\":\n line = re.sub(r'^(\\s*- )\\[[xX]\\]', r'\\1[ ]', line)\n\n result.append(line)\n\n # Add failure reason after the task line if failed\n if in_target_task and task_data and new_status == \"failed\" and reason:\n indent = \" \" * (task_indent // 2 + 1)\n # Check if next line already has a reason\n result.append(f\"{indent}- reason: {reason}\")\n in_target_task = False # Prevent adding reason multiple times\n\n return '\\n'.join(result)\n\n\ndef get_status_summary(tasks: list[Task]) -> dict:\n \"\"\"Get a summary of task statuses.\"\"\"\n\n summary = {\n \"total\": len(tasks),\n \"completed\": 0,\n \"pending\": 0,\n \"failed\": 0,\n \"blocked\": 0\n }\n\n completed_titles = {t.title for t in tasks if t.status == \"completed\"}\n\n for task in tasks:\n if task.status == \"completed\":\n summary[\"completed\"] += 1\n elif task.status == \"failed\":\n summary[\"failed\"] += 1\n elif task.status == \"pending\":\n # Check if blocked by dependencies\n deps_satisfied = all(dep in completed_titles for dep in task.dependencies)\n if deps_satisfied:\n summary[\"pending\"] += 1\n else:\n summary[\"blocked\"] += 1\n\n return summary\n\n\ndef cmd_next(args):\n \"\"\"Get the next task to execute.\"\"\"\n content = Path(args.file).read_text()\n tasks = parse_tasks_from_markdown(content)\n\n next_task = get_next_task(tasks)\n\n if args.json:\n if next_task:\n print(json.dumps({\n \"status\": \"found\",\n \"task\": asdict(next_task)\n }, indent=2))\n else:\n summary = get_status_summary(tasks)\n print(json.dumps({\n \"status\": \"no_tasks\",\n \"summary\": summary,\n \"message\": \"No pending tasks available\"\n }, indent=2))\n else:\n if next_task:\n print(f\"Next task: {next_task.title}\")\n print(f\"Priority: {next_task.priority} | Phase: {next_task.phase}\")\n if next_task.files:\n print(f\"Files: {', '.join(next_task.files)}\")\n if next_task.criteria:\n print(\"Criteria:\")\n for c in next_task.criteria:\n print(f\" - {c}\")\n else:\n print(\"No pending tasks available\")\n\n\ndef cmd_done(args):\n \"\"\"Mark a task as completed.\"\"\"\n file_path = Path(args.file)\n content = file_path.read_text()\n\n updated = update_task_status(content, args.task, \"completed\")\n file_path.write_text(updated)\n\n if args.json:\n print(json.dumps({\"status\": \"success\", \"task\": args.task, \"new_status\": \"completed\"}))\n else:\n print(f\"✅ Marked '{args.task}' as completed\")\n\n\ndef cmd_fail(args):\n \"\"\"Mark a task as failed.\"\"\"\n file_path = Path(args.file)\n content = file_path.read_text()\n\n updated = update_task_status(content, args.task, \"failed\", args.reason or \"\")\n file_path.write_text(updated)\n\n if args.json:\n print(json.dumps({\"status\": \"success\", \"task\": args.task, \"new_status\": \"failed\", \"reason\": args.reason}))\n else:\n print(f\"❌ Marked '{args.task}' as failed\")\n if args.reason:\n print(f\" Reason: {args.reason}\")\n\n\ndef cmd_status(args):\n \"\"\"Show status summary.\"\"\"\n content = Path(args.file).read_text()\n tasks = parse_tasks_from_markdown(content)\n summary = get_status_summary(tasks)\n\n if args.json:\n print(json.dumps({\n \"file\": args.file,\n \"summary\": summary,\n \"tasks\": [asdict(t) for t in tasks]\n }, indent=2))\n else:\n total = summary[\"total\"]\n completed = summary[\"completed\"]\n pct = round(completed / total * 100, 1) if total > 0 else 0\n\n print(f\"File: {args.file}\")\n print(f\"Progress: {completed}/{total} ({pct}%)\")\n print()\n print(f\" Completed: {summary['completed']}\")\n print(f\" Pending: {summary['pending']}\")\n print(f\" Blocked: {summary['blocked']}\")\n print(f\" Failed: {summary['failed']}\")\n\n # Show next task\n next_task = get_next_task(tasks)\n if next_task:\n print(f\"\\nNext: {next_task.title}\")\n\n\ndef cmd_list(args):\n \"\"\"List all tasks.\"\"\"\n content = Path(args.file).read_text()\n tasks = parse_tasks_from_markdown(content)\n\n if args.json:\n print(json.dumps([asdict(t) for t in tasks], indent=2))\n else:\n for task in tasks:\n status_icon = {\"completed\": \"✅\", \"failed\": \"❌\", \"pending\": \"⬜\"}.get(task.status, \"?\")\n print(f\"{status_icon} [{task.priority}] {task.title}\")\n\n\ndef main():\n parser = argparse.ArgumentParser(description=\"Markdown task manager\")\n subparsers = parser.add_subparsers(dest=\"command\", required=True)\n\n # next command\n next_parser = subparsers.add_parser(\"next\", help=\"Get next task\")\n next_parser.add_argument(\"--file\", required=True, help=\"Markdown file path\")\n next_parser.add_argument(\"--json\", action=\"store_true\", help=\"Output as JSON\")\n next_parser.set_defaults(func=cmd_next)\n\n # done command\n done_parser = subparsers.add_parser(\"done\", help=\"Mark task as completed\")\n done_parser.add_argument(\"--file\", required=True, help=\"Markdown file path\")\n done_parser.add_argument(\"--task\", required=True, help=\"Task title\")\n done_parser.add_argument(\"--json\", action=\"store_true\", help=\"Output as JSON\")\n done_parser.set_defaults(func=cmd_done)\n\n # fail command\n fail_parser = subparsers.add_parser(\"fail\", help=\"Mark task as failed\")\n fail_parser.add_argument(\"--file\", required=True, help=\"Markdown file path\")\n fail_parser.add_argument(\"--task\", required=True, help=\"Task title\")\n fail_parser.add_argument(\"--reason\", default=\"\", help=\"Failure reason\")\n fail_parser.add_argument(\"--json\", action=\"store_true\", help=\"Output as JSON\")\n fail_parser.set_defaults(func=cmd_fail)\n\n # status command\n status_parser = subparsers.add_parser(\"status\", help=\"Show status summary\")\n status_parser.add_argument(\"--file\", required=True, help=\"Markdown file path\")\n status_parser.add_argument(\"--json\", action=\"store_true\", help=\"Output as JSON\")\n status_parser.set_defaults(func=cmd_status)\n\n # list command\n list_parser = subparsers.add_parser(\"list\", help=\"List all tasks\")\n list_parser.add_argument(\"--file\", required=True, help=\"Markdown file path\")\n list_parser.add_argument(\"--json\", action=\"store_true\", help=\"Output as JSON\")\n list_parser.set_defaults(func=cmd_list)\n\n args = parser.parse_args()\n\n try:\n args.func(args)\n except FileNotFoundError:\n print(f\"Error: File not found: {args.file}\", file=sys.stderr)\n sys.exit(1)\n except Exception as e:\n print(f\"Error: {e}\", file=sys.stderr)\n sys.exit(1)\n\n\nif __name__ == \"__main__\":\n main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":13862,"content_sha256":"51f47e3129855560dccb88c906028ab4d48b0085439e0ccbc15540904643e11c"}],"content_json":{"type":"doc","content":[{"type":"heading","attrs":{"level":1},"content":[{"text":"Feature Pipeline","type":"text"}]},{"type":"paragraph","content":[{"text":"Execute implementation tasks directly from design documents. Tasks are managed as markdown checkboxes - no separate session files needed.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Quick Reference","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"# Get next task\npython3 scripts/task_manager.py next --file \u003cdesign.md>\n\n# Mark task completed\npython3 scripts/task_manager.py done --file \u003cdesign.md> --task \"Task Title\"\n\n# Mark task failed\npython3 scripts/task_manager.py fail --file \u003cdesign.md> --task \"Task Title\" --reason \"...\"\n\n# Show status\npython3 scripts/task_manager.py status --file \u003cdesign.md>","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Task Format","type":"text"}]},{"type":"paragraph","content":[{"text":"Tasks are written as markdown checkboxes in the design document:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"markdown"},"content":[{"text":"## Implementation Tasks\n\n- [ ] **Create User model** `priority:1` `phase:model`\n - files: src/models/user.py, tests/models/test_user.py\n - [ ] User model has email and password_hash fields\n - [ ] Email validation implemented\n - [ ] Password hashing uses bcrypt\n\n- [ ] **Implement JWT utils** `priority:2` `phase:model`\n - files: src/utils/jwt.py\n - [ ] generate_token() creates valid JWT\n - [ ] verify_token() validates JWT\n\n- [ ] **Create auth API** `priority:3` `phase:api` `deps:Create User model,Implement JWT utils`\n - files: src/api/auth.py\n - [ ] POST /register endpoint\n - [ ] POST /login endpoint","type":"text"}]},{"type":"paragraph","content":[{"text":"See ","type":"text"},{"text":"references/task-format.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/task-format.md","title":null}}]},{"text":" for full format specification.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Execution Loop","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":""},"content":[{"text":"LOOP until no tasks remain:\n 1. GET next task (task_manager.py next)\n 2. READ task details (files, criteria)\n 3. IMPLEMENT the task\n 4. VERIFY acceptance criteria\n 5. UPDATE status (task_manager.py done/fail)\n 6. CONTINUE","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Unattended Mode Rules","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"NO stopping","type":"text","marks":[{"type":"strong"}]},{"text":" for questions","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"NO asking","type":"text","marks":[{"type":"strong"}]},{"text":" for clarification","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Make autonomous decisions based on codebase patterns","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"If blocked, mark as failed and continue","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Status Updates","type":"text"}]},{"type":"paragraph","content":[{"text":"Completed task:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"markdown"},"content":[{"text":"- [x] **Create User model** `priority:1` `phase:model` ✅\n - files: src/models/user.py\n - [x] User model has email field\n - [x] Password hashing implemented","type":"text"}]},{"type":"paragraph","content":[{"text":"Failed task:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"markdown"},"content":[{"text":"- [x] **Create User model** `priority:1` `phase:model` ❌\n - files: src/models/user.py\n - [ ] User model has email field\n - reason: Missing database configuration","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Resume / Recovery","type":"text"}]},{"type":"paragraph","content":[{"text":"To resume interrupted work, simply run again with the same design file:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":""},"content":[{"text":"/feature-pipeline docs/designs/xxx.md","type":"text"}]},{"type":"paragraph","content":[{"text":"The task manager will find the first uncompleted task and continue from there.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Integration","type":"text"}]},{"type":"paragraph","content":[{"text":"This skill is typically triggered after ","type":"text"},{"text":"/feature-analyzer","type":"text","marks":[{"type":"code_inline"}]},{"text":" completes:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":""},"content":[{"text":"User: /feature-analyzer implement user auth\n\nClaude: [designs feature, generates task list]\n Design saved to docs/designs/2026-01-02-user-auth.md\n Ready to start implementation?\n\nUser: Yes / 开始实现\n\nClaude: [executes tasks via feature-pipeline]","type":"text"}]},{"type":"hr","attrs":{"markup":"---"}}]},"metadata":{"date":"2026-06-05","name":"feature-pipeline","author":"@skillopedia","source":{"stars":339,"repo_name":"happy-skills","origin_url":"https://github.com/notedit/happy-skills/blob/HEAD/skills/dev/feature-pipeline/SKILL.md","repo_owner":"notedit","body_sha256":"09f621fb15858bf58cfb774f6d874c2375c49edb8a7dd9663fd4be808a6c6948","cluster_key":"73c14dd2fc24e5bd4c3c445d3b480ce4e7b3b0f8aa367a4cc03ebac6a153ec4f","clean_bundle":{"format":"clean-skill-bundle-v1","source":"notedit/happy-skills/skills/dev/feature-pipeline/SKILL.md","attachments":[{"id":"af6d474b-40c7-5cef-82db-0a5a2ff6e19f","key":"uploads/10433ee7-ad12-4ae0-b34e-97553e46c6c8/af6d474b-40c7-5cef-82db-0a5a2ff6e19f/attachment.md","path":"references/task-format.md","size":2499,"sha256":"e23cd14885c880a4181f6d9b29f1417e7880523814948ac5793622a7317677d7","contentType":"text/markdown; charset=utf-8"},{"id":"c1816542-3e9c-5d69-a858-820581160c8c","key":"uploads/10433ee7-ad12-4ae0-b34e-97553e46c6c8/c1816542-3e9c-5d69-a858-820581160c8c/attachment.md","path":"references/workflow-guide.md","size":4043,"sha256":"dcd487d7a0592b6674956e7862c231745b93ac8483779c32a5d77f7ff22863c5","contentType":"text/markdown; charset=utf-8"},{"id":"5f59bff8-23f3-5da4-b980-a41f68e2ad7c","key":"uploads/10433ee7-ad12-4ae0-b34e-97553e46c6c8/5f59bff8-23f3-5da4-b980-a41f68e2ad7c/attachment.py","path":"scripts/task_manager.py","size":13862,"sha256":"51f47e3129855560dccb88c906028ab4d48b0085439e0ccbc15540904643e11c","contentType":"text/x-python; charset=utf-8"}],"bundle_sha256":"ef33c8fbd634a4da7a34376e5e94d801f932bcb31ecc40d09a7ef67741d3037d","attachment_count":3,"text_attachments":3,"attachment_storage":"skillopedia-attachments-v1","binary_attachments":0,"excluded_attachments":[]},"cluster_size":1,"skill_md_path":"skills/dev/feature-pipeline/SKILL.md","import_metadata":{"date":"2026-06-05","author":"@skillopedia","version":"v1","category":"documents-office","category_label":"Documents"},"exact_dupes_collapsed_into_this":0},"version":"v1","category":"documents-office","import_tag":"clean-skills-v1","description":"Execute implementation tasks from design documents using markdown checkboxes. Use when (1) implementing features from feature-analyzer output, (2) resuming interrupted work, (3) batch executing tasks. Triggers on 'start implementation', 'run tasks', 'resume'."}},"renderedAt":1782987917910}

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.