Fabric Council — Python

Use one Python fabric_exec. Pass payloads.task, a JSON array in payloads.roles, and a synthesis model key in payloads.synthesis; an empty payloads.synthesis keeps the session default. Choose 3–8 distinct non-empty roles with useful disagreements, not redundant reviewers. A role is a string or an object with role and optional model and thinking; a role without model runs on the session default. Label the outer display and every child. Use this template only for non-sensitive work: full reports are persisted in project-shared mesh, not private session storage. Do not supply credentials, secrets, or other material that must not be shared with other project sessions; use a plain agent without mesh parking for that work. Report secret locations and remediation, never secret values. This template does not automatically sanitize reports.

import asyncio
import json

async def ask(task, name, options=None, include_id=False):
    request = {"task": task, "name": name, "tools": ["read", "grep", "find", "ls"]}
    if options:
        request.update(options)
    result = await agents.run(request)
    if result["status"] != "completed":
        raise RuntimeError(result.get("error") or result["status"])
    report = result["value"] if result.get("value") is not None else result["text"]
    return {"report": report, "runId": result["id"]} if include_id else report

def text(value):
    return value.strip() if isinstance(value, str) else ""

def canonical(keys, needle, owner):
    key = keys.get(needle.lower())
    if key is not None:
        return key
    model_id = needle.lower().split("/")[-1]
    near = [candidate for candidate in keys.values() if candidate.lower().endswith("/" + model_id)]
    hint = " Did you mean: " + ", ".join(near[:8]) + "?" if near else ""
    raise ValueError("Council: model " + needle + " for " + owner + " is not in the model registry." + hint)

specs = json.loads(π.roles)
if not isinstance(specs, list):
    raise ValueError("Council roles must be a JSON array.")
seats = []
for spec in specs:
    entry = {"role": spec} if isinstance(spec, str) else spec
    if not isinstance(entry, dict):
        raise ValueError("Each council role must be a string or an object with role and optional model and thinking.")
    role = text(entry.get("role"))
    if not role:
        if isinstance(spec, str):
            continue
        raise ValueError("A council role object needs a non-empty role.")
    if [value for value in [entry.get("model"), entry.get("thinking")] if value is not None and not isinstance(value, str)]:
        raise ValueError("Council role " + role + " has a non-string model or thinking.")
    if text(entry.get("thinking")) and text(entry.get("thinking")) not in ["off", "minimal", "low", "medium", "high", "xhigh", "max"]:
        raise ValueError("Council role " + role + " has unknown thinking " + text(entry.get("thinking")) + ".")
    seat = {"role": role}
    if text(entry.get("model")):
        seat["model"] = text(entry.get("model"))
    if text(entry.get("thinking")):
        seat["thinking"] = text(entry.get("thinking"))
    earlier = [other for other in seats if other["role"] == role]
    if not earlier:
        seats.append(seat)
    elif earlier[0].get("model") != seat.get("model") or earlier[0].get("thinking") != seat.get("thinking"):
        raise ValueError("Council role " + role + " appears twice with different settings.")
if len(seats) < 3 or len(seats) > 8:
    raise ValueError("Council requires 3–8 distinct non-empty roles.")
synthesis_model = text(π.synthesis)
if synthesis_model or [seat for seat in seats if seat.get("model")]:
    # Match registry keys exactly before any agent runs, so a typo spends nothing.
    keys = {}
    for entry in await tools.models():
        keys[entry["key"].lower()] = entry["key"]
    for seat in seats:
        if seat.get("model"):
            seat["model"] = canonical(keys, seat["model"], "role " + seat["role"])
    if synthesis_model:
        synthesis_model = canonical(keys, synthesis_model, "synthesis")

async def review(seat):
    options = {}
    outcome = {"role": seat["role"]}
    if seat.get("model"):
        options["model"] = seat["model"]
        outcome["model"] = seat["model"]
    if seat.get("thinking"):
        options["thinking"] = seat["thinking"]
    try:
        report = await ask("Act as the " + seat["role"] + " council member. Independently analyze. Reports are persisted in project-shared mesh: never reproduce credentials, secret values, or private data; describe locations and remediation instead:\n" + π.task, seat["role"], options, include_id=True)
        outcome.update({"status": "completed", **report})
    except Exception as error:
        outcome.update({"status": "failed", "error": str(error)})
    return outcome

outcomes = await asyncio.gather(*[review(seat) for seat in seats])
completed = [item for item in outcomes if item["status"] == "completed"]
failures = [item for item in outcomes if item["status"] == "failed"]
coverage = {"requested": len(seats), "completed": len(completed)}
# Name the model each seat and the synthesis requested; an unpinned seat ran on the session default.
summary = {"coverage": coverage, "seats": [dict(seats[index], status=outcomes[index]["status"]) for index in range(len(seats))]}
if synthesis_model:
    summary["synthesisModel"] = synthesis_model
if not completed:
    return dict(summary, status="failed", failures=failures, result=None)
# Park completed reports in mesh before synthesis: an audit trail that survives this
# program, and full-text recovery when synthesis fails or a budget runs out.
# Use a host-assigned completed agent run ID: Monty has no native clock.
# ifVersion=0 keeps retries create-only instead of overwriting prior reports.
_parked = [
    {"role": item.get("role"), **({"model": item["model"]} if item.get("model") else {}), "report": item["report"]}
    for item in completed
]
reports = None
_park_error = None
_reports_base = "council/" + completed[0]["runId"]
for _attempt in range(4):
    _suffix = "" if _attempt == 0 else "-" + str(_attempt + 1)
    reports_key = _reports_base + _suffix + "/reports"
    try:
        await mesh.put(key=reports_key, value=_parked, ifVersion=0)
        reports = reports_key
        break
    except Exception as error:
        _park_error = error
        # Only a key collision retries with the next suffix; any other
        # storage failure surfaces immediately as the {"error": ...} object.
        if "compare-and-swap" not in str(error):
            break
if reports is None:
    reports = {"error": str(_park_error)}
if len(completed) == 1:
    return dict(summary, status="partial", failures=failures, result=completed[0]["report"], reports=reports, synthesisSkipped="Only one role completed; another agent adds no diversity.")
try:
    result = await ask("Synthesize only the completed reports into one decision. Reject unsupported claims, preserve material disagreements, and attribute concerns to roles. Never infer views of failed roles.\nTask:\n" + π.task + "\nReports:\n" + json.dumps(completed), "council synthesis", {"model": synthesis_model} if synthesis_model else None)
    return dict(summary, status="partial" if failures else "success", failures=failures, result=result, reports=reports)
except Exception as error:
    return dict(summary, status="partial", failures=failures, result=None, synthesisError=str(error), reports=reports, fallback=[{"role": item.get("role"), "status": "completed", "report": item["report"] if isinstance(reports, dict) else item["report"][:1200]} for item in completed])

Reserve at least the role count plus one agent call. Concurrency remains host-limited; usage settles afterward, not as a hard token reservation. Model keys must match a Pi registry key from tools.models() exactly, ignoring case; the program refuses a models.aliases name or a partial id, and checks every key before any agent runs. Whenever at least one seat completes, full role reports are parked to mesh under a create-only council/<run-id>/reports key (using the first completed agent's host-assigned run ID, with no native clock requirement) carried by every outcome as reports — a key collision parks under the next suffix (-2, -3, …) rather than overwriting: mesh.get it for audit or to retry a budget-starved synthesis with one plain agent, and mesh.delete it when no longer needed; on synthesis failure the fallback result carries report heads plus that mesh key, not the full texts; the synthesis prompt still embeds every completed report, so synthesis input grows with total report size. When mesh storage itself fails, reports is an {"error": ...} object instead of a key — never pass it to mesh.get or mesh.delete — and the synthesis-failure fallback then carries the full untruncated reports. A single completed role returns its full report directly as result without a synthesis call. Every result lists seats with each role's pinned model, thinking, and status, plus synthesisModel when one was pinned. These are the models the program requested; keeping a run on its pinned model is the host's job. Successful synthesis returns the compact decision, coverage, seats, failures, and reports (the mesh key, or the storage error object when parking failed); raw reports return on synthesis failure or when one role completes and synthesis is skipped. A partial result is useful evidence with named gaps, not permission for an automatic whole-council rerun. Retry only the failed role or synthesis when needed. Do not use a council for a lookup.