Design Task Decomposition Strategies for Complex Workflows

import anthropic
import asyncio

client = anthropic.Anthropic()

async def review_single_file(filename: str, content: str) -> dict:
    """PHASE 1: Per-file local analysis pass."""
    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=4096,
        messages=[{
            "role": "user",
            "content": f"""Review this file for LOCAL issues only.
Do NOT attempt cross-file analysis — focus strictly on this file.

File: {filename}
```python
{content}
```

Return JSON with:
- security_issues: list of {{issue, severity, line}}
- logic_errors: list of {{issue, description, line}}
- performance_issues: list of {{issue, impact, line}}
- summary: one-paragraph overview"""
        }]
    )
    return {"file": filename, "findings": response.content[0].text}


async def run_integration_pass(per_file_results: list[dict]) -> str:
    """PHASE 2: Cross-file integration analysis."""
    findings_text = "\n\n".join([
        f"=== {r['file']} ===\n{r['findings']}"
        for r in per_file_results
    ])

    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=4096,
        messages=[{
            "role": "user",
            "content": f"""You are given per-file analysis results for a codebase.
Identify CROSS-FILE issues that individual analysis missed:

{findings_text}

Identify:
1. Data flow issues (data from File A passed unsafely to File B)
2. Inconsistent error handling conventions across files
3. Dependency conflicts (circular imports, version mismatches)
4. Cross-file security vulnerabilities (e.g. auth bypass via API→DB path)
5. Contradictory findings (File A says X is safe; File B shows it isn't)

Return structured JSON with cross_file_issues[] and synthesis_summary."""
        }]
    )
    return response.content[0].text


async def code_review_pipeline(files: dict[str, str]) -> dict:
    """Full prompt-chaining code review: local passes → integration pass."""
    # PHASE 1: Run all per-file analyses in PARALLEL
    per_file_tasks = [
        review_single_file(fname, content)
        for fname, content in files.items()
    ]
    per_file_results = await asyncio.gather(*per_file_tasks)

    # PHASE 2: Integration pass uses per-file results as input
    integration_findings = await run_integration_pass(per_file_results)

    return {
        "per_file_findings": per_file_results,
        "integration_findings": integration_findings
    }

import anthropic
import json

client = anthropic.Anthropic()

def add_tests_to_legacy_codebase(repo_path: str):
    """Dynamic adaptive decomposition for open-ended test generation task."""

    # ─── PHASE 1: MAP STRUCTURE ──────────────────────────────────────────
    # Don't plan yet — discover first. The structure is unknown.
    mapping_response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=4096,
        messages=[{
            "role": "user",
            "content": f"""Map the structure of this codebase at: {repo_path}

Use the file system tools to explore. Produce a structural map containing:
- Complete module list with file counts
- Modules with ZERO test coverage (critical)
- Modules with partial coverage
- Dependency relationships between modules
- Entry points and critical paths

Return as JSON. Do NOT write any tests yet — mapping only."""
        }],
        tools=[/* file system tools */]
    )
    codebase_map = json.loads(mapping_response.content[0].text)

    # ─── PHASE 2: IDENTIFY HIGH-IMPACT AREAS ─────────────────────────────
    # Based on map, rank modules by impact before planning any tests.
    prioritization_response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"""Given this codebase structure:
{json.dumps(codebase_map, indent=2)}

Identify and rank the TOP 5 modules to test first, based on:
1. Zero current coverage (highest priority)
2. Critical path modules (payment, auth, data processing)
3. High dependency count (testing these blocks others)
4. Complexity (harder to test = more value in doing it right)

Return prioritized list with rationale for each. Do NOT write tests yet."""
        }]
    )
    priority_modules = json.loads(prioritization_response.content[0].text)

    # ─── PHASE 3: ADAPTIVE EXECUTION ─────────────────────────────────────
    # Generate tests for each priority module — plan adapts as dependencies found.
    generated_tests = []
    discovered_dependencies = []  # Grows as we explore

    for module in priority_modules["ranked_modules"]:
        test_response = client.messages.create(
            model="claude-opus-4-5",
            max_tokens=8096,
            messages=[{
                "role": "user",
                "content": f"""Generate comprehensive tests for: {module['name']}

Previously discovered dependencies (must mock these):
{json.dumps(discovered_dependencies, indent=2)}

If you discover NEW dependencies during test generation,
report them in a 'new_dependencies' field — they will be
added to the remaining testing plan.

Write pytest tests covering: happy path, edge cases, error conditions."""
            }],
            tools=[/* file read tools */]
        )

        result = json.loads(test_response.content[0].text)
        generated_tests.append(result["tests"])

        # KEY: Plan adapts as new dependencies are discovered
        if result.get("new_dependencies"):
            discovered_dependencies.extend(result["new_dependencies"])
            # These may be added to priority_modules for subsequent iterations

    return generated_tests