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
This commit is contained in:
Claude
2026-03-29 22:43:40 +00:00
commit 2d4ead1e3f
8 changed files with 703 additions and 0 deletions
+23
View File
@@ -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
+119
View File
@@ -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 <task>` | 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 <id>` | Mark a chunk as complete |
| `ctc show <id>` | 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
+3
View File
@@ -0,0 +1,3 @@
"""Claude Task Chunker - Break large tasks into efficient Claude Code sessions."""
__version__ = "0.1.0"
+106
View File
@@ -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
+215
View File
@@ -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 <chunk_id>")
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()
+110
View File
@@ -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,
}
+109
View File
@@ -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)
+18
View File
@@ -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*"]