commit 2d4ead1e3f9c98fe64d871b1e31dba1a1d2480cf Author: Claude Date: Sun Mar 29 22:43:40 2026 +0000 Add claude-task-chunker: break large tasks into efficient Claude Code sessions CLI tool that decomposes big coding tasks into self-contained chunks, each with a ready-to-paste prompt including codebase context, targeted instructions, and dependency tracking. Supports phase, component, and manual decomposition strategies. https://claude.ai/code/session_01Fzv8baXnEVVhnrffAb3Ucc diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..f89789c --- /dev/null +++ b/.gitignore @@ -0,0 +1,23 @@ +# Task chunker working files +.claude-chunks/ + +# Python +__pycache__/ +*.py[cod] +*$py.class +*.egg-info/ +dist/ +build/ +*.egg +.eggs/ + +# Virtual environments +.venv/ +venv/ +env/ + +# IDE +.idea/ +.vscode/ +*.swp +*.swo diff --git a/README.md b/README.md new file mode 100644 index 0000000..6814a5b --- /dev/null +++ b/README.md @@ -0,0 +1,119 @@ +# Claude Task Chunker (`ctc`) + +Break large coding tasks into self-contained chunks for efficient Claude Code sessions. Stop running out of usage mid-task. + +## The Problem + +You ask Claude Code to make a big change — add auth, refactor a module, migrate a database — and it runs out of usage halfway through. You're left with half-finished work and no easy way to pick up where you left off. + +## The Solution + +`ctc` decomposes large tasks into independent, self-contained chunks. Each chunk is a complete prompt you can paste into a fresh Claude Code session. It includes just enough context for Claude to work without needing the prior conversation. + +## Install + +```bash +pip install -e . +``` + +## Usage + +### 1. Plan — decompose a task + +```bash +ctc plan "Add OAuth2 authentication with login, logout, and RBAC" +``` + +This analyzes your codebase and creates a set of numbered chunks in `.claude-chunks/prompts/`. + +### 2. Execute — one chunk at a time + +```bash +# See what's next +ctc next + +# Or copy it straight to clipboard +ctc next -c + +# Paste the prompt into Claude Code and let it work +``` + +### 3. Track — mark chunks as done + +```bash +ctc done 1 +ctc status +``` + +### 4. Repeat until the full task is complete + +```bash +ctc next # get the next chunk +# paste into Claude Code +ctc done 2 # mark it complete +ctc next # and so on +``` + +## Decomposition Strategies + +### `phase` (default) +Breaks work into standard phases: Research & Plan → Core Implementation → Tests → Integration & Cleanup. Best for most tasks. + +```bash +ctc plan "Migrate from REST to GraphQL" -s phase +``` + +### `component` +Groups work by directory/module. Best when changes are spread across independent components. + +```bash +ctc plan "Add input validation everywhere" -s component +``` + +### `manual` +You define the chunks yourself in a text file, one per line. + +```bash +# chunks.txt +Set up database models for users and roles +Build registration and login API endpoints +Add JWT middleware and session management +Build role-based access control decorators +Add frontend login/logout pages +``` + +```bash +ctc plan "Add user auth" -s manual --chunks-file chunks.txt +``` + +## Commands + +| Command | Description | +|---------|-------------| +| `ctc plan ` | Decompose a task into chunks | +| `ctc status` | Show progress on current plan | +| `ctc next` | Print the next chunk's prompt | +| `ctc next -c` | Copy next chunk's prompt to clipboard | +| `ctc done ` | Mark a chunk as complete | +| `ctc show ` | Show a specific chunk's prompt | + +## How It Works + +1. **Analyzes** your codebase structure (languages, key files, directory layout) +2. **Decomposes** the task using the chosen strategy +3. **Generates** self-contained prompts with codebase context, targeted instructions, file lists, and verification steps +4. **Tracks** progress so you always know what's done and what's next + +Each generated prompt tells Claude Code: +- What the codebase looks like (summary, not full contents) +- Exactly what to do in this chunk +- Which files to read and modify +- What depends on what +- How to verify the work is correct + +## Tips + +- **Start with `phase` strategy** — it works well for most tasks +- **Use `manual` for complex tasks** — you know your codebase best; write chunks that make sense for your architecture +- **Keep chunks independent** — the less each chunk depends on others, the cleaner the handoff +- **Add `.claude-chunks/` to `.gitignore`** — it's working state, not source code diff --git a/claude_task_chunker/__init__.py b/claude_task_chunker/__init__.py new file mode 100644 index 0000000..631cbcb --- /dev/null +++ b/claude_task_chunker/__init__.py @@ -0,0 +1,3 @@ +"""Claude Task Chunker - Break large tasks into efficient Claude Code sessions.""" + +__version__ = "0.1.0" diff --git a/claude_task_chunker/analyzer.py b/claude_task_chunker/analyzer.py new file mode 100644 index 0000000..946c0b6 --- /dev/null +++ b/claude_task_chunker/analyzer.py @@ -0,0 +1,106 @@ +"""Codebase analyzer - gathers structure and context for task decomposition.""" + +import os +import subprocess +from dataclasses import dataclass, field +from pathlib import Path + + +@dataclass +class CodebaseInfo: + """Summary of a codebase's structure and key files.""" + root: Path + languages: dict[str, int] = field(default_factory=dict) # extension -> file count + total_files: int = 0 + total_lines: int = 0 + tree: list[str] = field(default_factory=list) # directory tree (paths) + key_files: list[str] = field(default_factory=list) # config, entry points, etc. + git_branch: str = "" + recent_commits: list[str] = field(default_factory=list) + + def summary(self) -> str: + lines = [f"Codebase: {self.root.name}"] + if self.git_branch: + lines.append(f"Branch: {self.git_branch}") + lines.append(f"Files: {self.total_files} | Lines: ~{self.total_lines}") + if self.languages: + top = sorted(self.languages.items(), key=lambda x: -x[1])[:5] + lines.append("Languages: " + ", ".join(f"{ext} ({n})" for ext, n in top)) + if self.key_files: + lines.append("Key files: " + ", ".join(self.key_files[:10])) + return "\n".join(lines) + + +IGNORE_DIRS = { + ".git", "node_modules", "__pycache__", ".venv", "venv", "dist", "build", + ".next", ".nuxt", "target", "vendor", ".tox", ".mypy_cache", ".pytest_cache", + "coverage", ".cache", "env", ".env", ".eggs", "*.egg-info", +} + +KEY_FILE_NAMES = { + "package.json", "pyproject.toml", "setup.py", "setup.cfg", "Cargo.toml", + "go.mod", "Makefile", "CMakeLists.txt", "Dockerfile", "docker-compose.yml", + "docker-compose.yaml", ".env.example", "requirements.txt", "tsconfig.json", + "webpack.config.js", "vite.config.ts", "vite.config.js", "next.config.js", + "tailwind.config.js", "CLAUDE.md", +} + + +def analyze_codebase(root: Path) -> CodebaseInfo: + """Walk the codebase and gather structural info.""" + info = CodebaseInfo(root=root.resolve()) + + # Git info + try: + info.git_branch = subprocess.check_output( + ["git", "rev-parse", "--abbrev-ref", "HEAD"], + cwd=root, stderr=subprocess.DEVNULL, text=True, + ).strip() + except (subprocess.CalledProcessError, FileNotFoundError): + pass + + try: + log = subprocess.check_output( + ["git", "log", "--oneline", "-10"], + cwd=root, stderr=subprocess.DEVNULL, text=True, + ).strip() + if log: + info.recent_commits = log.splitlines() + except (subprocess.CalledProcessError, FileNotFoundError): + pass + + # Walk files + languages: dict[str, int] = {} + total_lines = 0 + + for dirpath, dirnames, filenames in os.walk(root): + # Prune ignored directories + dirnames[:] = [d for d in dirnames if d not in IGNORE_DIRS] + + rel_dir = os.path.relpath(dirpath, root) + + for fname in filenames: + full = os.path.join(dirpath, fname) + rel = os.path.join(rel_dir, fname) if rel_dir != "." else fname + info.tree.append(rel) + info.total_files += 1 + + # Check key files + if fname in KEY_FILE_NAMES: + info.key_files.append(rel) + + # Language stats + ext = Path(fname).suffix.lower() + if ext: + languages[ext] = languages.get(ext, 0) + 1 + + # Line count (skip binary) + try: + with open(full, "r", errors="ignore") as f: + total_lines += sum(1 for _ in f) + except (OSError, UnicodeDecodeError): + pass + + info.languages = languages + info.total_lines = total_lines + return info diff --git a/claude_task_chunker/cli.py b/claude_task_chunker/cli.py new file mode 100644 index 0000000..f432e90 --- /dev/null +++ b/claude_task_chunker/cli.py @@ -0,0 +1,215 @@ +"""CLI interface for Claude Task Chunker.""" + +import argparse +import sys +from pathlib import Path + +from .analyzer import analyze_codebase +from .decomposer import TaskChunk, decompose_by_phase, decompose_by_component, STRATEGIES +from .tracker import save_plan, load_plan, mark_done, get_next_chunk, get_status_summary, TRACKER_DIR + + +def cmd_plan(args: argparse.Namespace) -> None: + """Decompose a task into chunks.""" + root = Path(args.project).resolve() + if not root.is_dir(): + print(f"Error: {root} is not a directory.", file=sys.stderr) + sys.exit(1) + + task = args.task + strategy = args.strategy + + print(f"Analyzing codebase at {root}...") + info = analyze_codebase(root) + summary = info.summary() + print(summary) + print() + + print(f"Decomposing task using '{strategy}' strategy...") + + if strategy == "phase": + raw_chunks = decompose_by_phase(task) + elif strategy == "component": + # Filter to source files only + source_exts = {".py", ".js", ".ts", ".tsx", ".jsx", ".go", ".rs", ".java", ".rb", ".cpp", ".c", ".h"} + source_files = [f for f in info.tree if Path(f).suffix.lower() in source_exts] + raw_chunks = decompose_by_component(task, source_files) + elif strategy == "manual": + # Read chunk descriptions from a file + if not args.chunks_file: + print("Error: --chunks-file is required with manual strategy.", file=sys.stderr) + sys.exit(1) + raw_chunks = _parse_manual_chunks(args.chunks_file, task) + else: + print(f"Error: Unknown strategy '{strategy}'.", file=sys.stderr) + sys.exit(1) + + chunks = [TaskChunk(**c) for c in raw_chunks] + plan_file = save_plan(root, task, chunks, summary) + + print(f"Created {len(chunks)} chunks:") + for c in chunks: + deps = f" (after: {', '.join(str(d) for d in c.depends_on)})" if c.depends_on else "" + print(f" {c.id}. {c.title}{deps}") + + print(f"\nPlan saved to: {plan_file}") + print(f"Prompts saved to: {root / TRACKER_DIR / 'prompts'}/") + print(f"\nTo start, copy the prompt from chunk_01.md into Claude Code.") + print(f"After each chunk, run: ctc done ") + + +def cmd_status(args: argparse.Namespace) -> None: + """Show current progress.""" + root = Path(args.project).resolve() + print(get_status_summary(root)) + + +def cmd_done(args: argparse.Namespace) -> None: + """Mark a chunk as completed.""" + root = Path(args.project).resolve() + chunk_id = args.chunk_id + + if mark_done(root, chunk_id): + print(f"Chunk {chunk_id} marked as done.") + print() + print(get_status_summary(root)) + else: + print(f"Error: Could not find chunk {chunk_id}.", file=sys.stderr) + sys.exit(1) + + +def cmd_next(args: argparse.Namespace) -> None: + """Show the next chunk's prompt, ready to paste.""" + root = Path(args.project).resolve() + plan = load_plan(root) + if not plan: + print("No active plan. Run `ctc plan` first.", file=sys.stderr) + sys.exit(1) + + nxt = get_next_chunk(root) + if not nxt: + print("All chunks are complete! Nothing left to do.") + return + + chunk = TaskChunk(**nxt) + prompt = chunk.prompt(plan["codebase_summary"], len(plan["chunks"])) + + if args.copy: + try: + import subprocess + process = subprocess.Popen( + _get_clipboard_cmd(), + stdin=subprocess.PIPE, + ) + process.communicate(prompt.encode()) + print(f"Chunk {chunk.id} prompt copied to clipboard!") + except (FileNotFoundError, OSError): + print("Could not copy to clipboard. Printing instead:\n") + print(prompt) + else: + print(prompt) + + +def cmd_show(args: argparse.Namespace) -> None: + """Show a specific chunk's prompt.""" + root = Path(args.project).resolve() + plan = load_plan(root) + if not plan: + print("No active plan. Run `ctc plan` first.", file=sys.stderr) + sys.exit(1) + + for c in plan["chunks"]: + if c["id"] == args.chunk_id: + chunk = TaskChunk(**c) + print(chunk.prompt(plan["codebase_summary"], len(plan["chunks"]))) + return + + print(f"Error: Chunk {args.chunk_id} not found.", file=sys.stderr) + sys.exit(1) + + +def _get_clipboard_cmd() -> list[str]: + """Get the platform-appropriate clipboard command.""" + import platform + system = platform.system() + if system == "Darwin": + return ["pbcopy"] + elif system == "Linux": + return ["xclip", "-selection", "clipboard"] + else: + return ["clip"] + + +def _parse_manual_chunks(filepath: str, task: str) -> list[dict]: + """Parse a simple text file where each non-empty line is a chunk description.""" + chunks = [] + path = Path(filepath) + if not path.exists(): + print(f"Error: {filepath} not found.", file=sys.stderr) + sys.exit(1) + + lines = [ln.strip() for ln in path.read_text().splitlines() if ln.strip() and not ln.startswith("#")] + + for i, line in enumerate(lines, 1): + chunks.append({ + "id": i, + "title": line, + "description": f"Part of: {task}\n\n{line}", + "depends_on": [i - 1] if i > 1 else [], + }) + return chunks + + +def main() -> None: + parser = argparse.ArgumentParser( + prog="ctc", + description="Claude Task Chunker - Break large tasks into efficient Claude Code sessions", + ) + parser.add_argument( + "-p", "--project", default=".", + help="Path to the project root (default: current directory)", + ) + sub = parser.add_subparsers(dest="command", required=True) + + # plan + p_plan = sub.add_parser("plan", help="Decompose a task into chunks") + p_plan.add_argument("task", help="Description of the full task") + p_plan.add_argument( + "-s", "--strategy", default="phase", + choices=["phase", "component", "manual"], + help="Decomposition strategy (default: phase)", + ) + p_plan.add_argument( + "--chunks-file", + help="File with one chunk description per line (for manual strategy)", + ) + + # status + sub.add_parser("status", help="Show progress on the current plan") + + # done + p_done = sub.add_parser("done", help="Mark a chunk as completed") + p_done.add_argument("chunk_id", type=int, help="ID of the completed chunk") + + # next + p_next = sub.add_parser("next", help="Show the next chunk prompt") + p_next.add_argument("-c", "--copy", action="store_true", help="Copy prompt to clipboard") + + # show + p_show = sub.add_parser("show", help="Show a specific chunk's prompt") + p_show.add_argument("chunk_id", type=int, help="ID of the chunk to show") + + args = parser.parse_args() + + commands = { + "plan": cmd_plan, + "status": cmd_status, + "done": cmd_done, + "next": cmd_next, + "show": cmd_show, + } + commands[args.command](args) + + +if __name__ == "__main__": + main() diff --git a/claude_task_chunker/decomposer.py b/claude_task_chunker/decomposer.py new file mode 100644 index 0000000..921e4fa --- /dev/null +++ b/claude_task_chunker/decomposer.py @@ -0,0 +1,110 @@ +"""Task decomposition engine - breaks large tasks into self-contained chunks.""" + +from dataclasses import dataclass, field + + +@dataclass +class TaskChunk: + """A self-contained unit of work for a single Claude Code session.""" + id: int + title: str + description: str + files_to_read: list[str] = field(default_factory=list) + files_to_modify: list[str] = field(default_factory=list) + depends_on: list[int] = field(default_factory=list) + verification: str = "" # how to verify this chunk is done + status: str = "pending" # pending | done | skipped + + def prompt(self, codebase_summary: str, total_chunks: int) -> str: + """Generate a ready-to-paste Claude Code prompt for this chunk.""" + lines = [ + f"# Task {self.id}/{total_chunks}: {self.title}", + "", + "## Context", + codebase_summary, + "", + "## What to do", + self.description, + "", + ] + + if self.files_to_read: + lines.append("## Files to review first") + for f in self.files_to_read: + lines.append(f"- `{f}`") + lines.append("") + + if self.files_to_modify: + lines.append("## Files to create or modify") + for f in self.files_to_modify: + lines.append(f"- `{f}`") + lines.append("") + + if self.depends_on: + deps = ", ".join(f"Task {d}" for d in self.depends_on) + lines.append(f"## Prerequisites") + lines.append(f"This task depends on: {deps}. Those changes should already be committed.") + lines.append("") + + if self.verification: + lines.append("## Verification") + lines.append(self.verification) + lines.append("") + + lines.extend([ + "## Instructions", + "- Focus ONLY on this task. Do not refactor or change unrelated code.", + "- Commit your changes when done with a clear commit message.", + "- If you encounter issues that block this task, document them and move on.", + ]) + + return "\n".join(lines) + + +# --- Built-in decomposition strategies --- + +def decompose_by_component(task: str, files: list[str]) -> list[dict]: + """Group files by directory (component/module) and create one chunk per group.""" + from collections import defaultdict + groups: dict[str, list[str]] = defaultdict(list) + + for f in files: + parts = f.split("/") + key = parts[0] if len(parts) > 1 else "root" + groups[key].append(f) + + chunks = [] + for i, (group, group_files) in enumerate(sorted(groups.items()), 1): + chunks.append({ + "id": i, + "title": f"{task} - {group}", + "description": f"Apply the following change to the `{group}` module/directory:\n\n{task}", + "files_to_modify": group_files, + }) + return chunks + + +def decompose_by_phase(task: str) -> list[dict]: + """Break a task into standard software engineering phases.""" + phases = [ + ("Research & Plan", f"Analyze the codebase and create a detailed plan for: {task}\n\nOutput a PLAN.md file listing:\n- Files that need to change\n- The order of changes\n- Any risks or dependencies\n- Test strategy"), + ("Core Implementation", f"Implement the core changes for: {task}\n\nFollow the plan in PLAN.md if it exists. Focus on the main logic, not tests or docs."), + ("Tests", f"Write tests for the changes made in: {task}\n\nReview the implementation that was already committed and add appropriate tests."), + ("Integration & Cleanup", f"Final integration pass for: {task}\n\nReview all changes made so far, fix any integration issues, update imports, and ensure everything works together. Run the existing test suite."), + ] + + return [ + { + "id": i, + "title": title, + "description": desc, + "depends_on": [i - 1] if i > 1 else [], + } + for i, (title, desc) in enumerate(phases, 1) + ] + + +STRATEGIES = { + "component": decompose_by_component, + "phase": decompose_by_phase, +} diff --git a/claude_task_chunker/tracker.py b/claude_task_chunker/tracker.py new file mode 100644 index 0000000..1b3a7a5 --- /dev/null +++ b/claude_task_chunker/tracker.py @@ -0,0 +1,109 @@ +"""Progress tracker - persists chunk status across sessions.""" + +import json +from dataclasses import asdict +from pathlib import Path + +from .decomposer import TaskChunk + +TRACKER_DIR = ".claude-chunks" + + +def get_project_dir(project_root: Path) -> Path: + d = project_root / TRACKER_DIR + d.mkdir(exist_ok=True) + return d + + +def save_plan(project_root: Path, task: str, chunks: list[TaskChunk], codebase_summary: str) -> Path: + """Save a decomposed plan to disk.""" + d = get_project_dir(project_root) + plan = { + "task": task, + "codebase_summary": codebase_summary, + "chunks": [asdict(c) for c in chunks], + } + plan_file = d / "plan.json" + plan_file.write_text(json.dumps(plan, indent=2)) + + # Also write individual prompt files for easy copy-paste + prompts_dir = d / "prompts" + prompts_dir.mkdir(exist_ok=True) + for chunk in chunks: + prompt_file = prompts_dir / f"chunk_{chunk.id:02d}.md" + prompt_file.write_text(chunk.prompt(codebase_summary, len(chunks))) + + return plan_file + + +def load_plan(project_root: Path) -> dict | None: + """Load an existing plan from disk.""" + plan_file = project_root / TRACKER_DIR / "plan.json" + if not plan_file.exists(): + return None + return json.loads(plan_file.read_text()) + + +def mark_done(project_root: Path, chunk_id: int) -> bool: + """Mark a chunk as completed.""" + plan = load_plan(project_root) + if not plan: + return False + + for chunk in plan["chunks"]: + if chunk["id"] == chunk_id: + chunk["status"] = "done" + plan_file = project_root / TRACKER_DIR / "plan.json" + plan_file.write_text(json.dumps(plan, indent=2)) + return True + return False + + +def get_next_chunk(project_root: Path) -> dict | None: + """Get the next pending chunk whose dependencies are satisfied.""" + plan = load_plan(project_root) + if not plan: + return None + + done_ids = {c["id"] for c in plan["chunks"] if c["status"] == "done"} + + for chunk in plan["chunks"]: + if chunk["status"] != "pending": + continue + deps = set(chunk.get("depends_on", [])) + if deps.issubset(done_ids): + return chunk + return None + + +def get_status_summary(project_root: Path) -> str: + """Get a human-readable status summary.""" + plan = load_plan(project_root) + if not plan: + return "No active plan found. Run `ctc plan` first." + + chunks = plan["chunks"] + done = sum(1 for c in chunks if c["status"] == "done") + total = len(chunks) + task = plan["task"] + + lines = [ + f"Task: {task}", + f"Progress: {done}/{total} chunks complete", + "", + ] + + for c in chunks: + status_icon = {"done": "[x]", "pending": "[ ]", "skipped": "[-]"}.get(c["status"], "[ ]") + deps = "" + if c.get("depends_on"): + deps = f" (after: {', '.join(str(d) for d in c['depends_on'])})" + lines.append(f" {status_icon} {c['id']}. {c['title']}{deps}") + + if done < total: + nxt = get_next_chunk(project_root) + if nxt: + lines.extend(["", f"Next up: Chunk {nxt['id']} - {nxt['title']}", + f" Prompt file: {TRACKER_DIR}/prompts/chunk_{nxt['id']:02d}.md"]) + + return "\n".join(lines) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..5bdd295 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,18 @@ +[build-system] +requires = ["setuptools>=68.0", "wheel"] +build-backend = "setuptools.backends._legacy:_Backend" + +[project] +name = "claude-task-chunker" +version = "0.1.0" +description = "Break large coding tasks into self-contained chunks for efficient Claude Code sessions" +readme = "README.md" +requires-python = ">=3.9" +license = {text = "MIT"} +dependencies = [] + +[project.scripts] +ctc = "claude_task_chunker.cli:main" + +[tool.setuptools.packages.find] +include = ["claude_task_chunker*"]