Openclaw Spawner

import asyncio
from spawn_openclaw import spawn_openclaw_async, is_error, read_result

async def main():
    payloads = [
        {
            "main_task_title": "Generate market research report",
            "progress_so_far": "Outline approved. Three sections assigned in parallel.",
            "sub_task": "Write 'Competitive Landscape' (600 words). Save to /tmp/section_competitive.md."
        },
        {
            "main_task_title": "Generate market research report",
            "progress_so_far": "Outline approved. Three sections assigned in parallel.",
            "sub_task": "Write 'Customer Segments' (600 words). Save to /tmp/section_customers.md."
        },
        {
            "main_task_title": "Generate market research report",
            "progress_so_far": "Outline approved. Three sections assigned in parallel.",
            "sub_task": "Write 'Market Trends' (600 words). Save to /tmp/section_trends.md."
        },
    ]

    summaries = await asyncio.gather(
        *[spawn_openclaw_async(p, depth=1, model="openai/gpt-4o", tools=my_tools) for p in payloads],
        return_exceptions=True,
    )

    result_paths = [
        "/tmp/section_competitive.md",
        "/tmp/section_customers.md",
        "/tmp/section_trends.md",
    ]

    for summary, path in zip(summaries, result_paths):
        if isinstance(summary, BaseException):
            print(f"Child raised exception for {path}: {summary}")
            continue
        if is_error(summary):
            print(f"Child failed for {path}:", summary)
            continue
        content = read_result(path)
        if content is None:
            print(f"WARNING: no result file at {path}")
        else:
            print(f"Merging {path} ({len(content)} chars)")
            # merge content into parent output ...

Parent:  ──[spawn child]──────────────────────────[merge point]──▶ continue
Child:            └──[work independently]──[write result file]──▶ done

import asyncio
from spawn_openclaw import spawn_openclaw_async, is_error, read_result

async def main():
    child_result_path = "/tmp/child_analysis.md"
    payload = {
        "main_task_title": "Refactor authentication module",
        "progress_so_far": (
            "Audit complete. Parent is now rewriting core auth logic. "
            "Child is assigned to analyse test coverage gaps in parallel."
        ),
        "sub_task": (
            f"Read `/tmp/audit_report.md`. Identify which functions lack test coverage. "
            f"Write a markdown report of gaps to `{child_result_path}`. "
            f"Include function name, file, and suggested test cases for each gap."
        ),
    }

    # Spawn — returns immediately, child runs in background
    child_task = asyncio.create_task(
        spawn_openclaw_async(payload, depth=1, model="anthropic/claude-opus-4-6", tools=my_tools)
    )

    # Parent does its OWN work right now
    await do_parent_work()

    # Merge point — collect child
    try:
        summary = await child_task
    except (ValueError, FileNotFoundError) as exc:
        print("Child raised configuration error:", exc)
        return
    if is_error(summary):
        print("Child failed:", summary)
    else:
        content = read_result(child_result_path)
        if content is None:
            print("WARNING: child did not write expected result file.")
        else:
            await integrate_child_output(content)


async def do_parent_work():
    pass  # replace with actual parent steps

async def integrate_child_output(text: str):
    print(f"Merging {len(text)} chars from child...")

try:
    summary = spawn_openclaw(payload, depth=1)
except (ValueError, FileNotFoundError) as exc:
    print("Configuration error:", exc)
    raise  # or handle

if is_error(summary):
    import json
    err = json.loads(summary)
    print("Runtime error:", err["error"])
    print("Partial results at:", err.get("partial_results"))  # may be None
    # decide: retry, fallback, abort parent
else:
    # success — read result files
    result = read_result("/tmp/some_output.md")
    if result is not None:
        # merge result ...
        pass
    else:
        print("WARNING: child did not write expected result file.")

SPAWN CHECKLIST
───────────────────────────────────────────
[ ] Sub-task is genuinely separable?
[ ] depth= will stay within MAX_DEPTH (default 3)?
[ ] Progress snapshot written (parent state frozen)?
[ ] main_task_title: ≤ 10 words, stable
[ ] progress_so_far: dense summary, no raw dumps
[ ] sub_task: self-contained, explicit result_path

STRATEGY SELECTION
[ ] Parent needs result before next step?        → Sequential (A)
[ ] Multiple children, all needed before merge?  → Parallel-gather (B)
[ ] Parent has its own work to do right now?     → Fire-and-forget (C)

AFTER EVERY SPAWN
[ ] is_error(summary) checked?
[ ] read_result(path) used (handles missing files safely)?
[ ] None result handled — don't merge silently?

FIRE-AND-FORGET EXTRAS
[ ] result_path agreed before spawning?
[ ] merge point placed after parent's own work?