fabric-rlm

Fabric Recursive Decomposition — Python

Use recursion for context size, not mere difficulty. Pass only the objective as payloads.task. Orient → delegate non-overlapping context-sized partitions → combine in one Python fabric_exec. Python calls agents.run(runner="pi", recursive=True, ...) directly for oversized partitions, not guest callback helpers. Plain leaves explicitly use recursive=False.

Context is an external variable

Keep source handles, partitions, and intermediate findings guest-local for this invocation. Children inspect paths or receive bounded slices, never the whole corpus. Guest bindings end with each call. For continuation across turns, persist JSON under root-scoped rlm/<rootId>/bindings/... mesh keys; send keys rather than values. For values above the mesh event limit use project-relative files plus digests. Mesh data is project-visible with no automatic TTL: no secrets; clean it up after completion. state is for claims and evidence, not scratch data.

import asyncio
import json

async def ask(task, name, recursive=False, output_schema=None):
    request = {"task": task, "name": name, "runner": "pi", "recursive": recursive, "tools": ["read", "grep", "find", "ls"]}
    if output_schema:
        request["schema"] = output_schema
    result = await agents.run(request)
    if result["status"] != "completed":
        raise RuntimeError(result.get("error") or result["status"])
    return result["value"] if result.get("value") is not None else result["text"]

partition_schema = {"type": "object", "properties": {"partitions": {"type": "array", "maxItems": 12, "items": {"type": "object", "properties": {"label": {"type": "string"}, "paths": {"type": "array", "items": {"type": "string"}}, "recursive": {"type": "boolean"}}, "required": ["label", "paths", "recursive"], "additionalProperties": False}}}, "required": ["partitions"], "additionalProperties": False}
scope = await ask("Partition relevant material into at most 12 non-overlapping context-sized groups. Set recursive=true only if a group cannot fit a child context.\nTask:\n" + π.task, "scope", output_schema=partition_schema)
proposed = scope["partitions"]
failures = []
candidates = []
for index in range(len(proposed)):
    partition = proposed[index]
    label = partition["label"].strip() or "partition-" + str(index + 1)
    paths = []
    for raw in partition["paths"]:
        value = raw.strip().replace("\\", "/")
        while value.startswith("./"):
            value = value[2:]
        while "//" in value:
            value = value.replace("//", "/")
        paths.append(value.rstrip("/"))
    invalid = not paths or any(not value or value in [".", "~"] or value.startswith("~/") or value.startswith("/") or value[1:3] == ":/" or ".." in value.split("/") for value in paths)
    if invalid:
        failures.append({"partition": label, "paths": paths, "status": "not_started", "error": "partition paths must be non-empty project-relative paths without '~' or '..'"})
        continue
    for value in paths:
        candidates.append([len(value.split("/")), len(value), index, label, value])
selected = []
grouped = {}
promoted = []
merged = []
for candidate in sorted(candidates):
    index, label, value = candidate[2], candidate[3], candidate[4]
    covered = None
    for entry in selected:
        if value == entry["path"] or value.startswith(entry["path"] + "/"):
            covered = entry
            break
    if covered:
        merged.append({"partition": label, "path": value, "coveredBy": covered["path"]})
        if proposed[index]["recursive"]:
            promoted.append(covered["index"])
        continue
    selected.append({"path": value, "index": index})
    if index not in grouped:
        grouped[index] = []
    grouped[index].append(value)
partitions = []
for index in range(len(proposed)):
    if index in grouped:
        partitions.append({"label": proposed[index]["label"].strip() or grouped[index][0], "paths": grouped[index], "recursive": proposed[index]["recursive"] or index in promoted})
normalization = {"proposed": len(proposed), "effective": len(partitions), "dispatched": 0, "mergedOverlaps": merged}
if not partitions:
    return {"status": "failed" if proposed else "success", "coverage": {"requested": len(proposed), "dispatched": 0, "completed": 0}, "failures": failures, "normalization": normalization, "result": None if proposed else "No relevant partitions were found."}
runnable = []
recursive_roots = 0
for partition in partitions:
    if partition["recursive"] and recursive_roots >= 2:
        failures.append({"partition": partition["label"], "paths": partition["paths"], "status": "not_started", "error": "recursive root limit reached"})
        continue
    if partition["recursive"]:
        recursive_roots += 1
    runnable.append(partition)

async def analyze(partition):
    task = "Analyze this bounded partition using paths as external context. Return compact evidence-backed findings.\nPartition: " + partition["label"] + "\nPaths:\n" + "\n".join(partition["paths"]) + "\nObjective:\n" + π.task
    try:
        finding = await ask(task, (("recurse " if partition["recursive"] else "analyze ") + partition["label"])[:50], recursive=partition["recursive"])
        return {"partition": partition["label"], "status": "completed", "finding": finding}
    except Exception as error:
        return {"partition": partition["label"], "paths": partition["paths"], "status": "failed", "error": str(error)}

outcomes = []
for offset in range(0, len(runnable), 4):
    batch = runnable[offset:offset + 4]
    settled = await asyncio.gather(*[analyze(partition) for partition in batch])
    outcomes.extend(settled)
    if all(item["status"] == "failed" for item in settled):
        failures.extend([{"partition": item["label"], "paths": item["paths"], "status": "not_started", "error": "not started after an all-failed batch"} for item in runnable[offset + len(batch):]])
        break
completed = [item for item in outcomes if item["status"] == "completed"]
failures.extend([item for item in outcomes if item["status"] == "failed"])
normalization["dispatched"] = len(outcomes)
coverage = {"requested": len(proposed), "dispatched": len(outcomes), "completed": len(completed)}
if not completed:
    return {"status": "failed", "coverage": coverage, "failures": failures, "normalization": normalization, "result": None}
if len(completed) == 1:
    return {"status": "partial" if failures else "success", "coverage": coverage, "failures": failures, "normalization": normalization, "result": completed[0]["finding"], "synthesisSkipped": "One partition completed"}
try:
    result = await ask("Synthesize only completed findings, reconcile duplicates and contradictions, drop unsupported claims, and never infer failed partitions.\nObjective:\n" + π.task + "\nFindings:\n" + json.dumps(completed), "combine")
    return {"status": "partial" if failures else "success", "coverage": coverage, "failures": failures, "normalization": normalization, "result": result}
except Exception as error:
    return {"status": "partial", "coverage": coverage, "failures": failures, "normalization": normalization, "result": None, "synthesisError": str(error), "fallback": completed}