Large projects are often overwhelming because it is difficult to decide how much to break them down into smaller steps.
It takes a big project and breaks it into smaller tasks while calculating the most efficient level of detail. It determines exactly how many sub-tasks are needed to balance work effectively.
It provides a clear way to find the right balance between having too many small tasks and one overwhelming project.
It was run in the sandbox and it failed. run output shows an error/traceback — the artifact does NOT run clean.
$ python3 recursive_task_decomposition_scorer_v2.py 📊 DEMO 1: Large project (complexity=5000, chunk_size=200) ======================================================================== RECURSIVE TASK DECOMPOSITION SCORER — REPORT ======================================================================== Root complexity : 5000.0 Chunk size : 200.0 Overhead per level : 1.00 ------------------------------------------------------------------------ Depth #Tasks AvgCplx Cost Benefit CBR CumCBR Stop ------------------------------------------------------------------------ 0 1 200.0 55.00 3360.00 0.0164 0.0164 ------------------------------------------------------------------------ Optimal depth : 0 Optimal CBR : 0.0164 Recommendation: OPTIMAL DEPTH = 0 (Cumulative CBR = 0.0164). Decompose to layer 0 for the best cost-benefit trade-off. Per-task complexity ~200.0 units. ======================================================================== 📊 DEMO 2: Tight chunking (complexity=300, chunk_size=150) ======================================================================== RECURSIVE TASK DECOMPOSITION SCORER — REPORT ==========================================
No screenshot — there is nothing working to show. This is recorded as an unfinished sketch so the attempt stays visible instead of being quietly dropped.
All of it — 207 lines, one file, standard library only.
#!/usr/bin/env python3
""" Recursive Task Decomposition Scorer v2
Adds Path Analysis feature to calculate cumulative cost and task counts across the entire tree at each depth level.
"""
import math
from dataclasses import dataclass, field
from typing import Optional, List
class TaskNode:
"""A node in the recursive task decomposition tree."""
id: str
total_complexity: float
chunk_size: float
overhead_per_level: float = 1.0
benefit_weight: float = 0.7
depth: int = 0
children: List['TaskNode'] = field(default_factory=list)
is_chunked: bool = False
chunk_count: int = 0
@dataclass
class DecompositionResult:
"""Result of scoring a single decomposition level."""
depth: int
task_count: int
per_task_complexity: float
cost: float
benefit: float
cost_benefit_ratio: float
cumulative_ratio: float
is_terminal: bool
cumulative_cost: float = 0.0 # Path Analysis addition
cumulative_tasks: int = 0 # Path Analysis addition
@dataclass
class ScoringReport:
"""Full report from running the scorer on a task tree."""
root_complexity: float
chunk_size: float
overhead_per_level: float
levels: List[DecompositionResult]
optimal_depth: int
optimal_cbr: float
recommendation: str
class TaskTreeModel:
def __init__(
self,
*,
total_complexity: float,
chunk_size: float,
overhead_per_level: float = 1.0,
benefit_weight: float = 0.7,
max_depth: int = 10,
):
self.root = TaskNode(
id="R",
total_complexity=float(total_complexity),
chunk_size=float(chunk_size),
overhead_per_level=float(overhead_per_level),
benefit_weight=float(benefit_weight),
depth=0,
)
self.max_depth = int(max_depth)
self._results: List[DecompositionResult] = []
self._recommendation: str = ""
def build(self) -> 'TaskTreeModel':
self.decompose(self.root, self.max_depth)
return self
def decompose(self, node: TaskNode, max_depth: int) -> None:
if node.depth >= max_depth:
return
if not self.should_decompose(node):
return
per_chunk = node.total_complexity / node.chunk_count
for i in range(node.chunk_count):
child = TaskNode(
id=f"{node.id}.{i + 1}",
total_complexity=per_chunk,
chunk_size=node.chunk_size,
overhead_per_level=node.overhead_per_level,
benefit_weight=node.benefit_weight,
depth=node.depth + 1,
)
node.children.append(child)
self.decompose(child, max_depth)
def should_decompose(self, node: TaskNode) -> bool:
if node.total_complexity <= node.chunk_size:
return False
node.chunk_count = math.ceil(node.total_complexity / node.chunk_size)
return node.chunk_count >= 2
def compute_cost(self, chunk_count: int, overhead_per_level: float) -> float:
coordination_penalty = chunk_count * (chunk_count - 1) * 0.05
return chunk_count * overhead_per_level + coordination_penalty
def compute_benefit(self, original_complexity: float, per_chunk_complexity: float, benefit_weight: float, depth: int) -> float:
raw_reduction = original_complexity - per_chunk_complexity
if raw_reduction <= 0:
return 0.0
decay = benefit_weight ** (depth + 1)
return raw_reduction * decay
def score(self) -> 'TaskTreeModel':
self._results = []
levels: dict[int, list[tuple[float, float, float]]] = {}
def walk(node: TaskNode):
if node.is_chunked:
cost = self.compute_cost(node.chunk_count, node.overhead_per_level)
per_task_cplx = node.total_complexity / node.chunk_count
benefit = self.compute_benefit(
node.total_complexity, per_task_cplx, node.benefit_weight, node.depth
)
levels.setdefault(node.depth, []).append((cost, benefit, per_task_cplx))
for child in node.children:
walk(child)
walk(self.root)
if not levels:
self._recommendation = "NO DECOMPOSITION - root task fits within chunk_size."
return self
cumulative_cost = 0.0
cumulative_tasks = 0
max_depth = max(levels.keys())
for d in range(max_depth + 1):
entries = levels.get(d, [])
if not entries:
continue
task_count = len(entries)
avg_cost = sum(e[0] for e in entries) / task_count
avg_benefit = sum(e[1] for e in entries) / task_count
avg_cplx = sum(e[2] for e in entries) / task_count
cumulative_cost += avg_cost * task_count
cumulative_tasks += task_count
cb_ratio = round(avg_cost / avg_benefit, 4) if avg_benefit > 0 else float('inf')
cum_ratio = round(cumulative_cost / cumulative_tasks, 4) if cumulative_tasks > 0 else float('inf')
self._results.append(
DecompositionResult(
depth=d,
task_count=task_count,
per_task_complexity=avg_cplx,
cost=avg_cost,
benefit=avg_benefit,
cost_benefit_ratio=cb_ratio,
cumulative_ratio=cum_ratio,
is_terminal=d == max_depth,
cumulative_cost=cumulative_cost,
cumulative_tasks=cumulative_tasks,
)
)
# Find optimal depth
optimal_depth = 0
optimal_cbr = float('inf')
for result in self._results:
if result.cumulative_ratio < optimal_cbr:
optimal_cbr = result.cumulative_ratio
optimal_depth = result.depth
self._recommendation = (
f"Optimal decomposition depth: {optimal_depth} \n" +
f"Cumulative cost at optimal depth: {optimal_cbr:.2f} \n" +
f"Total tasks at optimal depth: {self._results[optimal_depth].cumulative_tasks}"
)
return self
def report(self) -> ScoringReport:
return ScoringReport(
root_complexity=self.root.total_complexity,
chunk_size=self.root.chunk_size,
overhead_per_level=self.root.overhead_per_level,
levels=self._results,
optimal_depth=self._results.index(max(self._results, key=lambda x: x.cumulative_tasks)) if self._results else 0,
optimal_cbr=min(result.cumulative_ratio for result in self._results) if self._results else 0.0,
recommendation=self._recommendation,
)
if __name__ == "__main__":
# Example usage with path analysis
model = TaskTreeModel(
total_complexity=1000,
chunk_size=200,
overhead_per_level=1.0,
benefit_weight=0.7,
max_depth=3,
)
model.build()
model.score()
report = model.report()
print(f"Optimal depth analysis:\n{model._recommendation}\n\nDetailed levels:\n")
for level in report.levels:
print(f"Depth {level.depth}:\\n" +
f" Tasks: {level.task_count} (Cumulative: {level.cumulative_tasks})\\n" +
f" Cost: {level.cost:.2f} (Cumulative: {level.cumulative_cost:.2f})\n")