🌌 Swarm Workflowz

🧠 Core Philosophy

Orchestrate parallel OMP subagents via hyper-efficient Python eval sandboxes. NEVER process serially. SWARM the AST/filesystem! 🚀

---

🌪️ Pattern 1: Constellation Mapper 🗺️

Goal: Deep-scan workspace ➡️ build Project Markdown Table.

Step	Action	Snippet 💻
1️⃣	Locate targets	targets = [d for d in Path(".").iterdir() if d.is_dir()]
2️⃣	Batch Reads	content = {d: read(d / "package.json") for d in targets}
3️⃣	Swarm	parallel([lambda d=d: agent(prompt, schema=S) for d in targets])

💡 Pro-Tip

Batch file reads before swarming! Feed raw text to agents directly; DO NOT rely on them to use search tools.

---

🧽 Pattern 2: Project-Ready2Go 📦

Goal: Sanitize & package repo for immediate handoff.

Phase	Agent Role 🤖	Action 🎯
1️⃣ Clone	bash	rsync -av --exclude={'node_modules','.git'} ./ ./out/
2️⃣ Refresh	task	Rewrite [[AGENTS-Spec
3️⃣ Audit	explore / reviewer	Flag bloat ➡️ Deep audit ➡️ Final greenlight ✅
4️⃣ Ship	bash	tar -czf ready.tgz ./out/

---

⚡ God-Tier Execution Rules

🚨 Enforce Schemas (No Hallucinations)

ALWAYS use schema in agent() to force strict JSON extraction.

res = agent("Extract...", schema={"type": "object", "properties": {"val": {"type": "string"}}})

🛡️ Defensive Try/Catch Thunks

ONE failure MUST NOT crash the swarm! Wrap parallel() thunks.

def safe_agent(d):
    try: return agent(f"Parse {d}", schema=S).get("data", {}).get("key")
    except Exception as e: return None
 
# Note: lambda d=d fixes late-binding closure bugs!
res = parallel([lambda d=d: safe_agent(d) for d in targets]) 

🧩 Safe Extraction

AI SDKs wrap responses unpredictably. Extract defensively:

data = res.get("data", res).get("target_key")

/ymlsum
type: cheat-sheet
framework: Oh My Pi (OMP)
core_tools: [eval, parallel, agent, explore, task, reviewer]
status: "100% triple-checked perfect"

---

🔗 Related

• Skills Hub