Refine UltraPlan orchestration workflow

agentmain.py

@@ -14,10 +14,12 @@ except Exception: pass
 from ga import GenericAgentHandler, smart_format, get_global_memory, format_error, consume_file
 
 script_dir = os.path.dirname(os.path.abspath(__file__))
+BANNED_TOOLS = (['ask_user', 'start_long_term_update'] if '--no-user-tools' in sys.argv else [])
 def load_tool_schema(suffix=''):
     global TOOLS_SCHEMA
     TS = open(os.path.join(script_dir, f'assets/tools_schema{suffix}.json'), 'r', encoding='utf-8').read()
     TOOLS_SCHEMA = json.loads(TS if os.name == 'nt' else TS.replace('powershell', 'bash'))
+    TOOLS_SCHEMA = [t for t in TOOLS_SCHEMA if t.get('function', {}).get('name') not in BANNED_TOOLS]
 load_tool_schema()
 
 lang_suffix = '_en' if os.environ.get('GA_LANG', '') == 'en' else ''
@@ -205,6 +207,7 @@ if __name__ == '__main__':
     parser.add_argument('--verbose', action='store_true')
     parser.add_argument('--nobg', action='store_true')
     parser.add_argument('--nolog', action='store_true')
+    parser.add_argument('--no-user-tools', action='store_true')
     args, _unknown = parser.parse_known_args()
     _extra_args = dict(zip([k.lstrip('-') for k in _unknown[::2]], _unknown[1::2])) if _unknown else {}
 

assets/ga_ultraplan.py

@@ -1,15 +1,43 @@
-from contextlib import contextmanager
+from contextlib import contextmanager, redirect_stdout, redirect_stderr
 from concurrent.futures import ThreadPoolExecutor
 from time import time, sleep
-import html, os, subprocess, sys, tempfile, threading, webbrowser
+import html, io, json, os, re, subprocess, sys, tempfile, threading, traceback, urllib.request, webbrowser
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
-__all__ = ["phase", "parallel", "mapchain"]
+__all__ = ["plan", "phase", "parallel", "mapchain"]
 
-_T0 = time(); _phases = []; _phase_stack = []; _tasks = []; _current = "idle"; _events = []; _srv = None; _last = time(); _lock = threading.Lock()
-_RUN_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "temp", f"ultraplan_{int(_T0)}_{os.getpid()}")
+_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+_PORT = int(os.environ.get("GA_ULTRAPLAN_PORT", "47831"))
+_T0 = time(); _phases = []; _phase_stack = []; _tasks = []; _current = "idle"; _events = []; _srv = None; _last = time(); _lock = threading.Lock(); _exec_lock = threading.Lock()
+_TASK_SLUG = "task"; _FUNC_SEQ = 0; _PLANNED = False; _SESSION = None; _sessions = {}
+_RUN_DIR = os.path.abspath(os.environ.get("GA_ULTRAPLAN_RUNDIR", os.path.join(_ROOT, "temp", "ultraplan_default")))
 os.makedirs(_RUN_DIR, exist_ok=True)
 
+def _bind(rundir):
+    global _SESSION, _RUN_DIR, _phases, _phase_stack, _tasks, _current, _events, _FUNC_SEQ, _TASK_SLUG
+    key = os.path.abspath(rundir); os.makedirs(key, exist_ok=True)
+    s = _sessions.setdefault(key, {"rundir": key, "phases": [], "phase_stack": [], "tasks": [], "current": "idle", "events": [], "func_seq": 0, "task_slug": "task"})
+    _SESSION = key; _RUN_DIR = key; _phases = s["phases"]; _phase_stack = s["phase_stack"]; _tasks = s["tasks"]; _current = s["current"]; _events = s["events"]; _FUNC_SEQ = s["func_seq"]; _TASK_SLUG = s["task_slug"]
+    return s
+
+def _save_session():
+    if _SESSION in _sessions:
+        _sessions[_SESSION].update(current=_current, func_seq=_FUNC_SEQ, task_slug=_TASK_SLUG)
+
+def _need_plan():
+    if not _PLANNED: raise RuntimeError("call plan(rundir) as the first UltraPlan statement")
+
+def _slug(s):
+    s = re.sub(r"[^a-zA-Z0-9]+", "_", str(s)).strip("_").lower()
+    return s[:80] or "task"
+
+def _task_slug(path):
+    stem = os.path.splitext(os.path.basename(path or "task"))[0]
+    parts = [_slug(x) for x in re.split(r"[_\-]+", stem)]
+    stop = {"ultra", "ultraplan", "script", "boot", "build", "test", "debug", "verify", "explore", "reduce", "phase"}
+    parts = [p for p in parts if p and not p.isdigit() and p not in stop]
+    return "_".join(parts) or _slug(stem)
+
 def _note(s):
     global _last
     with _lock:
@@ -27,39 +55,78 @@ def _phase_lines(nodes, depth=0):
 
 def _page():
     with _lock:
-        lines = ["GA UltraPlan", "", f"current: {_current}", "", "phases:"]
-        lines += _phase_lines(_phases) or ["(none)"]
+        lines = ["GA UltraPlan"]
+        for key, s in _sessions.items():
+            lines += ["", f"== {os.path.basename(key) or key} ==", f"rundir: {key}", f"current: {s['current']}", "", "phases:"]
+            lines += _phase_lines(s["phases"]) or ["(none)"]
             lines += ["", "recent tasks:"]
-        lines += [f"{t['status']:<7} {t['desc']}" for t in _tasks[-12:]] or ["(none)"]
-        lines += ["", "events:", *_events[-30:]]
+            lines += [f"{t['status']:<7} {t['desc']}" for t in s["tasks"][-12:]] or ["(none)"]
+            lines += ["", "events:", *s["events"][-30:]]
+        if not _sessions: lines += ["", "(no sessions)"]
     return "<meta http-equiv=refresh content=1><pre>" + html.escape("\n".join(lines)) + "</pre>"
 
 class _H(BaseHTTPRequestHandler):
     def do_GET(self):
         b = _page().encode("utf-8"); self.send_response(200); self.send_header("Content-Type", "text/html; charset=utf-8"); self.end_headers(); self.wfile.write(b)
+    def do_POST(self):
+        global _TASK_SLUG, _PLANNED
+        if self.path != "/exec": self.send_response(404); self.end_headers(); return
+        n = int(self.headers.get("Content-Length", "0")); req = json.loads(self.rfile.read(n).decode("utf-8"))
+        out = io.StringIO(); err = io.StringIO(); rc = 0
+        with _exec_lock, redirect_stdout(out), redirect_stderr(err):
+            _bind(req["rundir"]); _note("exec: " + req.get("path", "<script>"))
+            cwd = os.getcwd(); old_env = os.environ.copy(); os.environ["GA_ULTRAPLAN_DAEMON"] = "1"; _PLANNED = False; _TASK_SLUG = req.get("task") or _task_slug(req.get("path")); _save_session()
+            try:
+                if req.get("cwd"): os.chdir(req["cwd"])
+                g = {"__name__": "__main__", "__file__": req.get("path", "<ultraplan>")}
+                exec(compile(req.get("code", ""), g["__file__"], "exec"), g, g)
+            except SystemExit as e:
+                rc = int(e.code or 0) if isinstance(e.code, int) else 1
+            except Exception:
+                rc = 1; traceback.print_exc()
+            finally:
+                _save_session(); os.chdir(cwd); os.environ.clear(); os.environ.update(old_env)
+        body = json.dumps({"returncode": rc, "stdout": out.getvalue(), "stderr": err.getvalue()}).encode("utf-8")
+        self.send_response(200); self.send_header("Content-Type", "application/json"); self.end_headers(); self.wfile.write(body)
     def log_message(self, *a): pass
 
-def _show():
+def _serve_daemon():
     global _srv
-    if _srv or os.environ.get("GA_ULTRAPLAN_HTML") == "0": return
-    _srv = ThreadingHTTPServer(("127.0.0.1", 0), _H)
-    url = f"http://127.0.0.1:{_srv.server_port}/"
+    sys.modules.setdefault("assets.ga_ultraplan", sys.modules[__name__])
+    _srv = ThreadingHTTPServer(("127.0.0.1", _PORT), _H); _srv.timeout = 60; url = f"http://127.0.0.1:{_PORT}/"
     print(f"[ultraplan] {url}", flush=True)
-    threading.Thread(target=_srv.serve_forever, daemon=True).start()
     if os.environ.get("GA_ULTRAPLAN_BROWSER") != "0": webbrowser.open(url)
-    def reap():
-        while True:
-            sleep(60)
-            if time() - _last > 3600: _srv.shutdown(); break
-    threading.Thread(target=reap, daemon=True).start()
+    while time() - _last < 3600: _srv.handle_request()
+
+def _ping():
+    try: urllib.request.urlopen(f"http://127.0.0.1:{_PORT}/", timeout=0.5).read(1); return True
+    except Exception: return False
+
+def _show():
+    if os.environ.get("GA_ULTRAPLAN_DAEMON") == "1" or os.environ.get("GA_ULTRAPLAN_HTML") == "0": return
+    if not _ping():
+        subprocess.Popen([sys.executable, __file__, "--daemon"], cwd=_ROOT, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env={**os.environ, "GA_ULTRAPLAN_DAEMON":"1"})
+        for _ in range(20):
+            if _ping(): break
+            sleep(0.25)
+
+def plan(rundir):
+    global _PLANNED
+    if _PLANNED: return
+    _PLANNED = True; _bind(rundir); _save_session()
+    if os.environ.get("GA_ULTRAPLAN_DAEMON") == "1": return
+    _show(); path = os.path.abspath(sys.argv[0]); code = open(path, encoding="utf-8").read()
+    data = json.dumps({"path": path, "cwd": os.getcwd(), "rundir": _RUN_DIR, "task": _task_slug(path), "code": code}).encode("utf-8")
+    r = urllib.request.urlopen(urllib.request.Request(f"http://127.0.0.1:{_PORT}/exec", data=data, headers={"Content-Type":"application/json"}), timeout=None)
+    resp = json.loads(r.read().decode("utf-8")); sys.stdout.write(resp.get("stdout", "")); sys.stderr.write(resp.get("stderr", "")); sys.exit(resp.get("returncode", 1))
 
 @contextmanager
 def phase(name, desc=""):
     global _current
-    _show(); t = time(); p = {"name": name, "desc": desc, "status": "run", "on": True, "children": [], "tasks": [], "ops": []}
+    _need_plan(); t = time(); p = {"name": name, "desc": desc, "status": "run", "on": True, "children": [], "tasks": [], "ops": []}
     with _lock:
         (_phase_stack[-1]["children"] if _phase_stack else _phases).append(p)
-        _phase_stack.append(p); _current = f"phase: {name}"
+        _phase_stack.append(p); _current = f"phase: {name}"; _save_session()
     print(f"[phase] {name}" + (f" - {desc}" if desc else ""), flush=True); _note(f"phase start: {name}")
     failed = False
     try:
@@ -74,7 +141,10 @@ def phase(name, desc=""):
             elif p in _phase_stack: _phase_stack.remove(p)
             if _phase_stack: _current = f"phase: {_phase_stack[-1]['name']}"
             else: _current = ("failed" if failed else "all phases done") + f"; last: {name} ({dt:.1f}s)"
-        print(f"[{status}] {name} ({dt:.1f}s)", flush=True); _note(f"phase {status}:  {name} ({dt:.1f}s)")
+            _save_session()
+        print(f"[{status}] {name} ({dt:.1f}s)", flush=True)
+        print("[next] Main agent must continue orchestration or stop; do not take over task work.", flush=True)
+        _note(f"phase {status}:  {name} ({dt:.1f}s)")
 
 def _task(desc, status="run"):
     with _lock:
@@ -93,14 +163,19 @@ def _fmt(x, data):
     return x.format(**data) if isinstance(x, str) else x
 
 def _subagent(desc, prompt=None, *, llm_no=0, timeout=3600):
+    global _FUNC_SEQ
     root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
-    with tempfile.NamedTemporaryFile("w", encoding="utf-8", suffix=".txt", prefix="ultra_", dir=_RUN_DIR, delete=False) as f:
+    _FUNC_SEQ += 1; path = os.path.join(_RUN_DIR, f"{_FUNC_SEQ:03d}_{_TASK_SLUG}_{_slug(desc)}.txt")
+    with open(path, "w", encoding="utf-8") as f:
         f.write(desc if prompt is None else prompt)
-    print(f"[subagent] {desc} -> {f.name}", flush=True); _note(f"agent: {desc}")
-    cmd = [sys.executable, os.path.join(root, "agentmain.py"), "--func", f.name, "--llm_no", str(llm_no), "--nobg", "--nolog"]
+    print(f"[subagent] {desc} -> {path}", flush=True); _note(f"agent: {desc}")
+    cmd = [
+        sys.executable, os.path.join(root, "agentmain.py"), "--func", path,
+        "--llm_no", str(llm_no), "--nobg", "--nolog", "--no-user-tools",
+    ]
     r = subprocess.run(cmd, cwd=root, text=True, capture_output=True, timeout=timeout)
     if r.returncode: raise RuntimeError(f"subagent failed: {desc}\n{r.stdout}\n{r.stderr}")
-    return os.path.splitext(f.name)[0] + ".out.txt"
+    return os.path.splitext(path)[0] + ".out.txt"
 
 def _run(task, data):
     task = task() if callable(task) else task
@@ -120,18 +195,21 @@ def _run(task, data):
 
 def parallel(tasks, max_workers=None, _label=None, **data):
     global _current
-    tasks = list(tasks); label = _label or f"parallel: {len(tasks)} tasks"
-    with _lock: _current = label
+    _need_plan(); tasks = list(tasks); label = _label or f"parallel: {len(tasks)} tasks"
+    with _lock: _current = label; _save_session()
     _op(label); _note(label)
     with ThreadPoolExecutor(max_workers=max_workers or min(3, len(tasks) or 1)) as ex:
         return list(ex.map(lambda t: _run(t, data), tasks))
 
 def mapchain(items, *steps, max_workers=None, **data):
     global _current
-    items = list(items); label = f"mapchain: {len(items)} items x {len(steps)} steps"
-    with _lock: _current = label
+    _need_plan(); items = list(items); label = f"mapchain: {len(items)} items x {len(steps)} steps"
+    with _lock: _current = label; _save_session()
     def run(x):
         for step in steps:
             d = {**data, "item": x, "previous": x}; x = _run(step(x) if callable(step) else step, d)
         return x
     return parallel([lambda x=x: run(x) for x in items], max_workers=max_workers, _label=label)
+
+if __name__ == "__main__" and "--daemon" in sys.argv:
+    _serve_daemon()

frontends/stapp.py

@@ -75,8 +75,10 @@ def build_prompt(objective):
     return f"""读取 {agent.log_path} 尾部，获取 agent 的最新输出。
 用户的 loop 诉求：<objective>{objective}</objective>
 判断该 agent 是否偷懒、是否真正完成诉求，用 <next_prompt></next_prompt> 输出要追加给它的指令：
-一般复述 objective，或不超过 10 字的督促，如：别停，继续 / 这就叫最优？你优化到位了吗 / 看我要求，你达成了吗 / 你好好看清楚 / 你能不能看看记忆
-只输出 <next_prompt>…</next_prompt>。"""
+一般复述 objective，或不超过 10 字的**督促**，如：别停，继续 / 这就叫最优？你优化到位了吗 / 看我要求，你达成了吗 / 你好好看清楚 / 你能不能看看记忆 / 把关键发现和阶段性成果落盘，然后继续
+不允许促进 agent 停止或代替宣告任务完成，只允许催促不要对原任务进行评价，特别**禁止**“任务已完成，结束”这种让agent结束的指令，你的任务是让agent继续loop而非停止。
+只输出 <next_prompt>…</next_prompt>，若需要停止则不要输出此tag。
+"""
 
 @st.cache_resource
 def get_controller():
@@ -386,11 +388,11 @@ elif st.session_state.get('display_queue') is not None:
 # ── 空闲自主行动：fragment 定时检测，替代 launch.pyw 的 idle_monitor ──
 @st.fragment(run_every=timedelta(minutes=1))
 def _idle_checker():
-    if st.session_state.get('loop_enabled'):                       # 循环回程:取controller回填
+    if st.session_state.get('loop_enabled'):
         b = get_controller()
         if b['ready']:
             b['ready'] = False
-            if b['out']: st.session_state['_inject_prompt'] = b['out']
+            if b['out'] and '停止循环' not in b['out']: st.session_state['_inject_prompt'] = b['out']
             else: st.session_state.loop_enabled = False
             st.rerun(scope="app")
         return

memory/procmem_scanner.py

@@ -34,14 +34,37 @@ k32.VirtualQueryEx.restype = SIZE_T
 k32.ReadProcessMemory.argtypes = [PHANDLE, LPCVOID, LPVOID, SIZE_T, ctypes.POINTER(SIZE_T)]
 k32.ReadProcessMemory.restype = ctypes.wintypes.BOOL
 
-def is_hex_pattern(pattern):
-    clean = pattern.replace(" ", "").replace("??", "")
-    return all(c in "0123456789abcdefABCDEF" for c in clean) and (len(clean) % 2 == 0 or "??" in pattern)
+import re
 
-def build_rules(pattern, mode='auto'):
+# Regex to expand YARA (n) jumps to explicit ?? chains (YARA 4.5.4 bug)
+_RE_JUMP = re.compile(r'\(\s*(\d+)\s*\)')
+
+def expand_yara_jumps(hex_pattern):
+    """Expand (n) → ?? repeated n times, e.g. '90 ( 32 ) 00' → '90 ?? ??...?? 00'"""
+    def _repl(m):
+        return ' '.join(['??'] * int(m.group(1)))
+    return _RE_JUMP.sub(_repl, hex_pattern)
+
+def is_hex_pattern(pattern):
+    """Detect hex patterns like '90 ( 32 ) 00' or '90 ?? 00'"""
+    clean = pattern.replace(" ", "").replace("??", "")
+    # Also remove parenthesized jump counts like (32)
+    clean = _RE_JUMP.sub('', clean)
+    return all(c in "0123456789abcdefABCDEF" for c in clean) and len(clean) % 2 == 0
+
+def build_rules(pattern, mode=None):
+    if hasattr(pattern, 'match'): return pattern
+    mode = mode or ('auto' if isinstance(pattern, str) else 'yara')
+    if mode in ('yara',):
+        try:
+            return yara.compile(source=str(pattern))
+        except yara.SyntaxError:
+            raise  # user-provided full YARA rule, don't mess with it
+    # hex mode or auto
     use_hex = (mode == 'hex') or (mode == 'auto' and is_hex_pattern(pattern))
     if use_hex:
-        rule_text = f'rule CustomSearch {{ strings: $h = {{ {pattern.strip()} }} condition: $h }}'
+        hex_body = expand_yara_jumps(pattern.strip())
+        rule_text = f'rule CustomSearch {{ strings: $h = {{ {hex_body} }} condition: $h }}'
     else:
         escaped = pattern.replace('\\', '\\\\').replace('"', '\\"')
         rule_text = f'rule CustomSearch {{ strings: $s = "{escaped}" ascii wide condition: $s }}'
@@ -60,7 +83,7 @@ def format_llm_context(data, offset, base_addr, length=64):
         "hit_pos": offset - start
     }
 
-def scan_memory(pid, pattern, context_size=256, mode='auto', llm_mode=False):
+def scan_memory(pid, pattern, context_size=256, mode=None, llm_mode=False):
     rules = build_rules(pattern, mode)
     h_proc = k32.OpenProcess(0x0400 | 0x0010, False, pid)
     if not h_proc:
@@ -87,9 +110,10 @@ def scan_memory(pid, pattern, context_size=256, mode='auto', llm_mode=False):
                 data = buf.raw[:read.value]
                 for match in rules.match(data=data):
                     for inst in match.strings:
-                        offset = inst.instances[0].offset
-                        matched_data = inst.instances[0].matched_data
                         base = mbi.BaseAddress if mbi.BaseAddress else 0
+                        for instance in inst.instances:  # ITERATE ALL instances, not just [0]
+                            offset = instance.offset
+                            matched_data = instance.matched_data
                             if llm_mode:
                                 results.append(format_llm_context(data, offset, base, length=context_size))
                             else:

memory/ultraplan_sop.md

@@ -1,453 +1,181 @@
 # GA UltraPlan SOP
-
-UltraPlan is GA's controller-driven protocol for work that is too broad, adversarial, or uncertain for a single linear agent pass.
-
-The core model is **not** "one script equals one MapReduce". A script is a **bounded, file-backed MapReduce program**: it may contain multiple phases, multiple fan-out/fan-in passes, and multiple per-item chains, but it should only cover the portion of the task that is currently plannable. When the script finishes, UltraPlan is not automatically finished. The main agent must read the outputs, reduce them into task state, and decide whether to stop, act directly, ask the user, or launch the next UltraPlan script.
-
-```text
-main agent controller
-  -> write/run a bounded UltraPlan script
-  -> read file-backed outputs
-  -> reduce into shared state / decision
-  -> stop, direct edit, ask, or write the next script
-```
-
-## Non-negotiable principles
-
-1. **Script completion is not task completion.** All phases being done only means the currently planned program returned evidence.
-2. **Plan only the currently plannable horizon.** If later work depends heavily on unknown results, stop after producing those results and let the main agent regenerate the next script.
-3. **Everything important is file-backed.** Subagents return `.out.txt` files (the library manages their output dir); shared state lives in explicit files you create under `temp/`, never in the code tree. The exact directory does not matter; the wiring does.
-4. **The main agent owns global reduction.** Subagents produce local evidence, attempts, critiques, or transforms. The main agent owns cross-run state, prioritization, final decisions, and user-facing claims.
-5. **Use independent perspectives on purpose.** Parallel agents should not be redundant copies unless redundancy itself is the experiment.
-6. **No silent completeness claims.** If coverage is bounded, sampled, timed out, or intentionally skipped, say so.
-7. **Scale to the request, not to ambition.** A casual ask gets a few agents and single-pass judgement; "exhaustively audit / be thorough" earns a larger pool and 3-5-vote adversarial verification. Heavy fan-out is opt-in by the request, never inferred.
-
-## Workflow script contract
-
-A GA UltraPlan script is plain Python using `../assets/ga_ultraplan.py`:
-
+## 1. Protocol: start and continue
+### What this is
+UltraPlan is Python-scripted multi-agent orchestration. The main agent designs phases, prompts, fan-out/fan-in, and stop/continue decisions; subagents do task-facing work.
+### Opt-in only
+Start UltraPlan only when the user explicitly says `ultraplan`, `UltraPlan`, or `ultraplan mode`. If not opted in, do not start it; at most mention it is available.
+### First move
+Once opted in, the next substantive action is writing and running the first script.
+Before the first `plan(...)`, do not inspect source, tests, logs, imports, file listings, pages, or APIs for the task itself.
+Allowed pre-launch work: record objective/constraints, confirm cwd is GA `temp/`, write the minimal script.
+### File and cwd contract
+Scripts are plain Python files under GA `temp/`; run them with cwd = `temp/`.
+Reference repo files from `temp/`, e.g. `../assets/...`; never place UltraPlan scripts in the repo code tree.
+Every script starts with the real API contract and a shared artifact directory:
 ```python
-from assets.ga_ultraplan import phase, parallel, mapchain
+import os, sys
+sys.path.append("..")
+from assets.ga_ultraplan import plan, phase, parallel, mapchain
+RUN_DIR = os.path.abspath("ultraplan_<stable_slug>")
+plan(RUN_DIR)
+ARTIFACT_DIR = RUN_DIR
 ```
+`plan(...)` must be the first UltraPlan statement; defining `RUN_DIR` before it is allowed. If import/plan fails, diagnose only cwd/path/import/daemon startup, not the user task.
+### Same-plan continuation
+For one user objective, every later script reuses the exact same `RUN_DIR` and `plan(RUN_DIR)`.
+Round 2/3/etc. are new scripts under the same plan/work directory, not new plans. Continuation changes only phases/prompts/archetype.
+A finished script is not proof the task is finished: read reducer/report paths, then answer, ask, apply a completed result, or launch the next same-plan script.
+### Delegation boundary
+Do not solve the task outside UltraPlan. Do not perform task discovery, implementation, review, or verification in main chat.
+The main agent may read outputs only to supervise and decide the next script.
+Exactly one agent is the UltraPlan orchestrator for one objective. The orchestrator may read this SOP; ordinary workers must not be told to read UltraPlan SOP, start UltraPlan, design phases, or delegate.
+If the orchestrator itself is a subagent, give only objective, constraints, output budget, and permission to use UltraPlan; do not paste SOP, prescribe phases, or tell it which SOP files to read. It chooses context reads.
+Worker prompts are job tickets, not mini-SOPs: role, exact scope, inputs, allowed/forbidden actions, evidence, short output shape, stop condition.
+Every worker prompt must include the boundary when relevant: `Do not start UltraPlan. Do not delegate. If decomposition is needed, report blocker only.`
 
-Run scripts from the GA code root so `assets` imports correctly. Express orchestration only through `phase()`, `parallel()`, and `mapchain()`; design prompts, shared files, and reduction steps around those primitives.
+## 2. Core mental model
+### Why orchestrate
+Assume a strong single executor can handle long documents, complex code, and coherent multi-file edits. Do not orchestrate merely because work looks large.
+Orchestration is mainly omission control: missing items, angles, hypotheses, evidence, checks, or residual improvements. Hunt is the special hard-search case: one direct attempt may hit many dead ends before a viable proof/root cause/solution appears.
+### Three decisions
+1. Problem class: Explore, Sweep, Hunt, or Improve.
+2. Omission risk: unknown-list discovery or known-list ownership.
+3. Topology: one executor, parallel width, phase/loop depth, pipeline, barrier, reducer.
+### Parallel is only for omission control
+Use parallel in exactly two cases:
+- Unknown-list discovery: the item list is not known; split by meaningful search lenses, paths, evidence sources, representations, failure modes, or counterexamples. Evidence sources may include local code, logs/tests, live reproduction, user artifacts, and web/Google research when external ecosystem knowledge may reveal known issues, API limits, prior incidents, or platform constraints.
+- Known-list ownership: the item list is known, independent, and AI-sized; assign ownership so no item is skipped.
+Do not parallelize coherent execution. If the task is clear, bounded, coherent, and in one capable agent's comfort zone, use one executor.
+### Width vs depth
+Parallel width: independent angles/items may surface different omissions.
+Phase/loop depth: later search depends on earlier findings, reduction, verification, or dead ends. Use phases/loops for find -> dedupe/rank -> verify/refute -> search residuals until dry.
+### Choose by main risk
+Explore: space unknown; risk is missing angles/items.
+Sweep: known independent list; risk is missing items/status.
+Hunt: cause/solution/proof unknown or hard; risk is wrong path/dead ends.
+Improve: existing artifact; risk is residual defects/opportunities after execution.
+Design/Integrator are support moves: design contracts prevent divergent parallel output; integrators restore global coherence after parallel work.
 
-Put the script file itself and every artifact it writes under `temp/` (or a user-specified area), never in the code tree (root, `assets/`, `memory/`). Running from the root is only for the import to resolve — it is not a license to drop the script there. If a `temp/` script needs the import, add the root to `sys.path` inside it; do not move the script into the code tree to make the import convenient.
+## 3. Tool semantics and output discipline
+`phase(name, desc="")` is visible structure. Name the current archetype and reducer boundary.
+`parallel(tasks, max_workers=None, **data)` runs independent tasks and returns result paths in input order. Default concurrency is engine-chosen; omit `max_workers` in examples unless there is a real reason.
+Task forms: tuple/list `(desc, prompt)` or dict with `desc`, `prompt`, `data`, `llm_no`, `timeout`.
+Subagent calls return `.out.txt` paths. Later prompts should reference paths and tell workers to read/tail only what they need.
+`mapchain(items, step1, step2, ...)` runs steps sequentially per item and items concurrently. `{item}` is original item; `{previous}` is the prior step result path.
+A `parallel(...)` between stages is a barrier. Use it only when the next stage needs cross-result dedupe, ranking, shared context, or early-exit. If each item can continue independently, use `mapchain`.
+Workers return plain text, not JSON by default, but it must be reducer-readable: stable IDs, evidence paths/quotes, verdict/status, risk, next action.
+Brevity rule: be as short as practical. Main chat reports only status, blocker, next action. Workers/reducers/verifiers include necessary evidence but no padding.
+Forbid filler: no background essay, copied prompt, SOP recap, chain-of-thought prose, unsupported impression, or vague `done`.
+Reducers compare rather than concatenate: accept, reject, dedupe, rank, expose contradictions, state coverage bounds, and recommend stop/continue/next archetype.
 
-Each task can be one of:
+## 4. Prompt contract
+A worker prompt must be executable without follow-up.
+State: role, exact scope, input paths/items, artifact directory, allowed sources/tools, evidence standard, concise output shape, stop condition, and exclusions. If web/Google search is allowed, say so explicitly; require URLs/source names, distinguish sourced facts vs local evidence vs hypotheses, and map every external finding to a task hypothesis, discriminator, mitigation, or verification step.
+Tell workers what not to do when overlap is harmful. Tell verifiers whether to confirm, refute, reproduce, compare, or inspect local formatting.
+Prefer file paths over pasted long context. Give only the state needed for that worker.
+If a worker creates or edits files, require saving them under `ARTIFACT_DIR` and returning paths.
+If an operation is risky or irreversible, the prompt must stop before doing it unless the user already approved it.
 
+## 5. Archetypes
+### Explore
+Use Explore when the space is unknown and choosing one path too early would bias the task.
+Fan out by lenses, not by fake dependencies: architecture, failure modes, data/evidence sources, constraints, user intent, external web/official/forum evidence, reproduction route, counterexample route, test surface, style risk. Use web search only when outside knowledge may change the map; forbid generic background research.
+Each explorer returns lens, covered area, findings/frontiers, evidence, unknowns, and dead ends.
+Reducer builds the map: accepted facts, rejected claims, promising frontiers, missing lenses, contradictions, and next archetype.
+Stop Explore when the reducer can name a bounded Execute, Hunt, Improve, or Sweep.
+Research/report tasks often start with material collection plus Explore of different research paths; save gathered material as files, then synthesize a report and Improve it. In engineering/debugging tasks, web research is a collector lens feeding a reducer, not the final artifact unless the user asked for a research report.
+### Hunt
+Use Hunt for uncertain root cause, high-stakes claim validation, or hard solution/proof search.
+Typical flow: collect evidence surface -> synthesize facts/timeline/contradictions -> generate diverse hypotheses/approaches -> rank by evidence/value/cost/verifiability -> verify selected candidates.
+Fan out by non-overlapping blades or evidence sources: local code/static path, logs/errors/tests, reproduction behavior, recent changes, dependency edges, external web/official/forum evidence, constraints, weird angles, alternate representation, counterexamples. Use web research when known ecosystem/platform failures may be missing from local evidence; it must return cited mechanisms and discriminators, not a background essay.
+Each hunter returns candidate/approach ID, evidence, confidence, why distinct, why plausible, how to verify, and dead ends.
+Verification becomes Sweep only after there is a known independent hypothesis list.
+If all attempts fail, record rejected paths, exclude repeats, change blades/representation, and Hunt again. Dead ends are progress.
+### Improve
+Use Improve when there is an existing artifact to fix, simplify, optimize, rewrite, polish, or decide among alternatives.
+Improve is an outer loop, not one edit: find opportunities/residual risks -> reduce/prioritize -> execute selected change -> verify/test -> search residuals -> repeat.
+Do not start Improve with mechanical fixed lenses. Ask what omissions matter for this artifact, then choose search lenses only when each lens can uniquely find something.
+Default execution is one AI executor for small/coupled/coherent work. Sweep only when there is a known independent item list; Hunt if cause/option is unknown.
+Example: for a single-file/coherent rewrite, use one executor, then run real tests (see Verification shapes: discover existing tests/demos/CLI usage or build minimal real ones, not import-only smoke), then optionally use unknown-list discovery to find remaining regressions, style issues, or missed simplifications.
+After execution, verify. If coverage is uncertain, use phase/loop depth: find plausible untested failures -> dedupe/rank -> verify/refute -> search residuals until dry. Use parallel width only for distinct discovery lenses.
+For important changes, use adversarial verify: ask workers to refute the result, not merely agree.
+Stop only when no material improvement remains or remaining items are tiny/unsafe.
+### Sweep
+Use Sweep when the item list is known, items are truly independent, and coverage/status matters.
+Classic case: download games A/B/C/D. Each item has its own search/download/verify route; parallel ownership prevents forgetting D and isolates blockers.
+Use `mapchain` for per-item inspect -> act -> verify when each item can progress without global waiting.
+Each item report includes item ID, action/result, evidence, status, unresolved risk, skipped condition, and blocker.
+Reducer reports total, covered, omitted, failed, accepted findings, rejected findings, and coverage bounds.
+Do not call every large dataset a Sweep. 12,000 correlated rows for analysis usually need one data-analysis executor/script, not 12,000 AI workers. Sweep is for AI-sized independent items, often few-to-dozens; for huge independent batches, sample/pilot then script/shard with explicit bounds.
+If several items fail similarly, reduce failures and switch to Hunt for common cause; if one item is hard, make that item a Hunt.
+
+## 6. Composition rules and edge cases
+### Design before parallel construction
+If independent artifacts require shared style/terminology/format, first Explore/Design a compact contract, then Sweep construction, then one integrator pass.
+Example: add Troubleshooting sections to 12 independent docs pages. Contract first; page workers then write under contract; integrator unifies style and catches hallucinations.
+But high-coherence artifacts with small per-unit edits usually stay single-executor. Example: add one conclusion sentence to each non-title slide in a 40-slide PPT; narrative continuity is global, so one executor writes. Sweep is suitable only for local checks such as missing sentence, overflow, or layout errors.
+### Verification shapes
+Verify is not always Sweep. For single coherent artifacts, verification often uses unknown-list discovery: find possible problems, reduce them, verify/refute, then search residuals until no material issue remains.
+Use parallel verification only when distinct lenses can find different omissions. Otherwise use one verifier plus real tests.
+Decouple verification lenses by evidence source, not by sub-checklists of one method. Splitting one static read into "check API", "check parity", "check side effects" is fake parallelism: same method, same files, overlapping output. Real independent sources are usually: static analysis (read code, no run), real execution (run actual tests), and quality-vs-intent (does it meet the task's goal, e.g. simpler/cleaner). Open a parallel lens only when its evidence source is genuinely different.
+Real execution must be genuine, not import-only smoke. Optimize for finding real breakage, not for finishing cheap. The runner first discovers existing entry points (test suites, example/demo scripts, README/CLI usage, callable public APIs) and runs the relevant ones; if none cover the change, it builds minimal but real tests that exercise the changed behavior. Record exact commands and stdout/stderr/stack. Never report pass for behavior that was not actually run; mark it blocked with the concrete reason (e.g. needs live window/GPU/network) and what would unblock it.
+Sweep verification fits known local independent checks: each file has required logging format, each slide has no overflow, each downloaded game opens.
+High-stakes claims use Hunt-style validation: collect evidence, generate alternatives, verify/refute, and block confident answers if coverage is weak.
+### Research/report shape
+Research is usually not a primitive archetype. Use collection/exploration to gather materials, parallel paths for distinct search strategies, a reducer/synthesizer for the report, then Improve to remove synthesis scars, gaps, weak evidence, and style problems.
+Final chat should summarize what was produced and where files/materials are, not paste huge gathered content.
+### Multi-round continuation
+Do not write one giant script when the next phase depends on reduced results.
+After each script, read only reducer/report outputs needed to decide: answer, ask, apply completed result, or launch the next same-plan script.
+The next script's archetype comes from the reducer: Explore if the map is still unknown, Hunt for candidates, Improve for chosen artifact, Sweep for known items, Verify for high-stakes claims.
+Never restart outside UltraPlan or rename the plan because the first script finished; rename only for a different user objective.
+
+## 7. Scale, failure, and bounds
+Scale to the request. Quick check uses small fan-out; comprehensive audit uses broader blades, stronger verification, and explicit coverage bounds.
+Prefer engine-chosen concurrency. More agents are worse when prompts overlap; improve decomposition before tuning execution knobs.
+Use `timeout` for risky or slow probes and require workers to report partial progress.
+If you sample, top-N, time-box, skip retries, exclude a subsystem, or hit a tool failure, make the bound visible in reducer output and final answer.
+If a worker fails, inspect its `.out.txt` or error path, then retry with narrower scope, longer timeout, different tool, or different archetype. Do not repeat a failed prompt unchanged.
+If reducers expose contradictions, launch targeted verification to resolve conflicts with evidence.
+If coverage is too weak, do not answer confidently; run another same-plan script or ask the user to choose cost/coverage.
+
+## 8. Classic patterns
+Use these as recognition anchors, not rigid templates:
+1. Improve existing artifact/code: Improve loop. If execution is coherent, one executor changes it; real tests follow (discover existing tests/demos/CLI usage or build minimal real ones, not import-only smoke); use unknown-list discovery only to find residual regressions, missed simplifications, style issues, or weak tests; repeat until dry.
+2. Root cause / unsafe conclusion: Hunt. Collect evidence first (single collector if narrow; parallel collectors only for distinct evidence sources) -> synthesize -> find hypotheses/counterexamples -> verify/refute -> record dead ends and continue if unresolved.
+3. Many known independent deliverables: Sweep. Example: download A/B/C/D games. Each item gets ownership because sequential work often forgets items; methods may differ per item. Reducer tracks status/blockers.
+4. Large correlated data: not Sweep per row. 12000 rows needing analysis is one coherent data-analysis execution; Sweep only independent AI-sized subsets or residual problem cases.
+5. Research/report: parallel only for distinct search paths/sources because materials may be missed; synthesize with one writer/integrator; Improve then searches evidence gaps, synthesis scars, style problems, and missing perspectives.
+6. Simple coherent code change across modest files: one executor may do it; add Sweep only for known local checks such as per-file log format, then optional residual discovery for style/tests.
+7. Single file or high-coherence artifact verification: not Sweep. Use tests and unknown-list problem discovery; use parallel only if distinct lenses can find different omissions.
+8. Design-then-Sweep construction: when items are independent but style must match, first Explore/Design a contract, then Sweep item work, then one integrator pass.
+9. PPT/narrative conclusion edits: usually one executor for coherence; Sweep may check local layout/format only, not write each page when cross-page flow matters.
+
+## 9. Minimal shapes
 ```python
-("Short display name", "Prompt template with {item}, {previous}, ...")
+BOUNDARY = "Do not start UltraPlan. Do not delegate. If decomposition is needed, report blocker only."
+ART = f"Save any artifacts under {ARTIFACT_DIR}; return paths."
 
-{
-    "desc": "Short display name with {item}",
-    "prompt": "Prompt template",
-    "data": {"extra": "values"},
-    "llm_no": 0,
-    "timeout": 3600,
-}
+with phase("Improve coherent artifact", "single executor -> real tests -> residual search"):
+    result = parallel([("Executor", f"Apply the focused change. Keep coherence. Run real tests: find existing tests/demos/CLI usage and run them, or build minimal real ones; record commands and output; do not claim pass for unrun behavior. {ART} Return evidence/blockers. Be concise. {BOUNDARY}")])[0]
 
-lambda: ("Short display name", "Prompt generated lazily at runtime")
+with phase("Find residuals", "only if coverage is uncertain"):
+    # Fill only meaningful lenses; leave empty for one verifier.
+    residual_lenses = []
+    residuals = parallel(residual_lenses) if residual_lenses else parallel([
+        ("Verifier", f"Inspect {result}. Find blockers/residuals only. {ART} Return evidence and stop/continue. Be concise. {BOUNDARY}")])
+
+with phase("Reduce/decide", "dedupe, verify/refute, continue/stop"):
+    next_move = parallel([("Reducer", f"Use {result} and {residuals}. Return accepted/rejected, artifact paths, evidence, next action. Be concise. {BOUNDARY}")])[0]
+
+with phase("Hunt", "evidence -> hypotheses -> verification plan"):
+    evidence = parallel(evidence_collectors)  # each collector prompt includes ART and BOUNDARY
+    hypotheses = parallel(hypothesis_blades) # each hunter prompt includes ART and BOUNDARY
+    ranked = parallel([("Reducer", f"Use {evidence} and {hypotheses}. Rank candidates and verification steps. Return artifact paths. Be concise. {BOUNDARY}")])[0]
+
+with phase("Sweep known independent items", "per-item ownership and status"):
+    reports = mapchain(items,
+        ("Inspect {item}", "Inspect only {item}. Save artifacts under {artifact_dir}; return paths, ID, evidence, action, risk. Be concise. " + BOUNDARY),
+        ("Act/verify {previous}", "Use {previous}. Save artifacts under {artifact_dir}; return paths, ID, status, evidence, blocker. Be concise. " + BOUNDARY),
+        artifact_dir=ARTIFACT_DIR)
 ```
-
-A task returns the path of its subagent output file, normally `.../temp/ultra_xxx.out.txt`. Treat that path as the workflow value. Do not rely on subagents producing valid JSON unless a later Python step explicitly validates and repairs it.
-
-## Shared workspace: give independent agents just enough shared context
-
-Subagents are independent. A prompt alone is often too narrow: it lacks the evolving global view and encourages duplicated work. For any serious UltraPlan run, put the few shared files it needs under a task folder like `temp/<task-slug>/` (or a user-specified area), never in the code tree. The library auto-creates its own per-run output dir for `.out.txt` files; you do not manage or pass it.
-
-Do not impose a fixed schema or fixed filenames. Name shared files by the role they play in this task: context, item list, prior attempts, target, criteria, frontier, accepted findings. One context file may be enough; a Sweep may need an item list; a Hunt may need prior attempts; an Improve run may need the target plus criteria.
-
-Rules:
-
-- The controller creates and refreshes shared files before launching a script.
-- Worker prompts say exactly which shared file(s) to read and which target item, lens, or shard they own.
-- Workers may write scratch notes only when explicitly assigned a unique file. Avoid concurrent appends to the same file.
-- Reducers read worker `.out.txt` files plus the relevant shared files; they do not infer global truth from one worker.
-- Between scripts, the main agent updates whichever notes are useful: tried attempts, accepted findings, rejected claims, remaining frontier, coverage bounds, or changed priorities.
-- A worker shares no context with you or its siblings; its prompt must stand alone. Hand it pointers, not payloads: pass a **file path** for anything large or unstructured and let it Read; pass small structured data (an item, a lens, a config) **directly** in the prompt. Never spill a long blob into the prompt, and never write a side-channel file just to smuggle a value a direct argument could carry. Its returned text IS the data, not a message to a human.
-- Pure data shaping — dedup, flatten, filter, sort, count — is plain controller code, not an agent. Spend an agent only where judgement is required.
-
-## Primitive semantics
-
-### `phase(name, desc="")`
-
-Use `phase` as a visible boundary in the HTML phase tree and as a semantic boundary in the program. A phase should describe what is being learned, attempted, or reduced.
-
-```python
-with phase("Explore target surface", "map unknown areas before choosing attacks"):
-    maps = parallel([...])
-```
-
-Good phase names are verbs tied to the archetype: `Hunt candidates`, `Verify hits`, `Critique improvement angles`, `Sweep item list`, `Reduce frontier`.
-
-### `parallel(tasks, max_workers=None, **data)`
-
-Run independent tasks concurrently and return output-file paths in input order. This is a barrier: the next Python statement runs only after all tasks complete.
-
-Use `parallel` when a reduction truly needs the full set:
-
-- brainstorming diverse candidates before choosing the best;
-- independent sharp critics attacking the same target from different angles;
-- partitioned sweep results that must be deduplicated;
-- a judge/reducer comparing mutually exclusive approaches.
-
-Do not insert a barrier merely because stages have different names. If each item can move through its stages independently, use `mapchain`.
-
-### `mapchain(items, *steps, max_workers=None, **data)`
-
-Run each item through all steps independently. There is no cross-item barrier between stages: item A can be in verification while item B is still being inspected.
-
-Each step receives the current value as `{item}` and `{previous}`. Initially this is the original item; after each subagent task it becomes that subagent's `.out.txt` path.
-
-When a later step consumes a prior `.out.txt`, instruct it to **tail** that file, not read it, unless the full transcript matters. `.out.txt` files contain the previous worker's full transcript; the useful conclusion is usually near the end.
-
-```python
-reports = mapchain(
-    items,
-    ("Inspect {item}", "Read this run's task folder context, then inspect assigned item: {item}. Output evidence-backed findings."),
-    ("Verify {previous}", "Tail {previous}. Try to refute or reproduce the finding with tools."),
-)
-```
-
-Use `mapchain` by default for per-item multi-stage work. It keeps item context local, avoids artificial barriers, and prevents the main agent from hand-managing intermediate maps.
-
-## Controller cycle: UltraPlan is recursive MapReduce, not a single run
-
-A full UltraPlan often needs several scripts. Each script is one bounded program over the currently knowable horizon; the main agent is the controller between programs.
-
-Use this cycle:
-
-1. **Scout cheaply in the main agent.** Read the request, obvious files, target pages, logs, or prior notes. Do not spawn blind fan-out before identifying the target object.
-2. **Create or refresh only the needed shared notes.** Use task-shaped files such as a target/context note, criteria, attempt history, item list, or current findings. Do not assume a universal schema.
-3. **Choose the current archetype.** Hunt, Improve, Explore, or Sweep. A script may contain multiple phases and multiple MapReduce barriers, but it should have one dominant purpose.
-4. **Write and run the bounded script.** Encode the control flow up front using `phase`, `parallel`, and `mapchain`.
-5. **Read outputs physically.** Read the final reducer and any important intermediate `.out.txt` files. Do not trust file names, status lines, or success messages.
-6. **Reduce into controller state.** Update the useful shared notes: what was tried, what survived verification, what remains unknown, what coverage is bounded, and what changed priority.
-7. **Decide next action.** Stop and answer, edit directly, ask the user, or launch a new UltraPlan script with a new archetype.
-
-This is the key distinction: all phases in one script being complete only ends that script's MapReduce program. It does not prove the overall task is complete.
-
-You do not need the full shape before the *task* — only before the next *orchestration step*. Stay in the loop: run one well-scoped fan-out, read its result, then decide the next phase, rather than committing the entire expand→reduce→verify graph up front. Several small scripts read end-to-end beat one ambitious script that guessed wrong.
-
-## Four recursive archetypes
-
-These are not a fixed catalog of examples. They are four first-principles control shapes that can transform into one another. A serious task often moves from one archetype to another across scripts.
-
-### 1. Hunt: goal-directed search until one viable success is found
-
-Use when the task has a clear success condition and many possible routes. You need broad ideation, non-duplicate attempts, and fast validation.
-
-Examples: find a working exploit path, find any viable implementation approach, find one source that proves a claim, find a way around an integration blocker.
-
-Control shape:
-
-1. Expand the candidate frontier from multiple independent lenses.
-2. Deduplicate candidates against the recorded attempt history or other relevant shared notes.
-3. Try candidates in parallel or with `mapchain(candidate -> attempt -> verify)`.
-4. Stop early only if success is physically verified; otherwise update the frontier and launch the next hunt wave.
-
-Prompt requirements:
-
-- Each hunter must state its lens and avoid repeating recorded attempts.
-- Each attempt must specify exact success evidence.
-- Verifiers should default to "not proven" unless they can reproduce or directly inspect the evidence.
-
-### 2. Improve: optimize a target by finding sharp defects or upgrade angles
-
-Use when an object exists and the goal is to make it better. Independence is critical: issue-finders must be pointed, skeptical, and non-colluding, not generic reviewers.
-
-Examples: improve a design, harden a plan, optimize prompts, reduce latency, refine an SOP, improve UX, raise code quality.
-
-Control shape:
-
-1. Freeze the current target object in a concrete shared note or file reference.
-2. Launch independent critics with distinct sharp lenses: correctness, missing cases, simplicity, performance, security, user value, maintainability, originality, verification.
-3. Reduce critiques into a prioritized issue list with evidence.
-4. Launch a second script to sweep fixes or design specific remedies.
-5. Verify that each accepted improvement actually improves the target and does not break constraints.
-
-Prompt requirements:
-
-- Critics should be adversarial and specific, not balanced reviewers.
-- A critique is useful only if it names the exact target fragment, failure mode, and improvement direction.
-- The reducer must reject duplicate, vague, or unsupported criticism.
-
-### 3. Explore: determine what exists, what matters, or whether the target is real
-
-Use when the objective itself is uncertain: you do not yet know the landscape, relevant files, possible mechanisms, or whether a phenomenon exists.
-
-Examples: understand an unfamiliar codebase area, investigate a bug class, discover whether a site exposes an API, map a research topic, determine whether a feature already exists.
-
-Control shape:
-
-1. Partition the unknown space into independent lenses or regions.
-2. Have explorers gather evidence, not final answers.
-3. Reduce into a map of knowns, unknowns, contradictions, and promising next questions.
-4. Convert the next step into Hunt, Improve, or Sweep once the target becomes concrete.
-
-Prompt requirements:
-
-- Explorers must report scope covered and scope not covered.
-- Reducers must distinguish evidence, inference, and speculation.
-- Exploration should produce a frontier, not pretend to close the task.
-
-### 4. Sweep: enumerate and process every item without duplication or omission
-
-Use when the main risk is coverage: every file, row, endpoint, finding, document, or case must be considered.
-
-Examples: per-file migration planning, verify every reported issue, process every source, audit every route, update every usage site.
-
-Control shape:
-
-1. Build an item list and validate the enumeration method.
-2. Use `mapchain` for per-item work: inspect -> transform/plan -> verify.
-3. Reduce item outputs into a coverage table.
-4. Run a gap-checker against the item list and any relevant attempt/coverage notes.
-5. If gaps remain, launch another sweep over the missing set.
-
-Prompt requirements:
-
-- Each worker owns exactly one item or shard.
-- Each output must include item ID, action taken/proposed, evidence, and unresolved risk.
-- Reducers must report coverage count and omissions explicitly.
-
-## Archetype transformations
-
-The four archetypes recursively call each other:
-
-- An **Improve** run may discover unknown areas, causing an **Explore** sub-run.
-- An **Explore** run may identify a success route, causing a **Hunt**.
-- A **Hunt** that finds a candidate may require **Improve** to harden it.
-- Any accepted plan may require **Sweep** to apply or verify it across every item.
-- A **Sweep** may expose failures, returning to **Hunt** for workarounds or **Improve** for fixes.
-
-Do not force all transformations into one script. If the next archetype depends on the reducer's result, stop the script, update shared state, and write a new script.
-
-## Script templates by archetype
-
-These templates teach **how to decompose the task**, not how to wire files. Read each one for its lens choices and why those lenses are independent. File wiring is deliberately reduced to a one-line task-folder header (`ctx`, `attempts`, ...); every worker is assumed to read the relevant shared file, so prompts state only the *thinking*, not `Read X and Y`. Run from the GA code root. Keep scripts short enough that the controller can read outputs and decide the next archetype.
-
-### Hunt template
-
-The decomposition idea: **cut the candidate space with several non-overlapping blades** so the union covers more ground than any single hunter would, then dedupe, attempt, and verify.
-
-```python
-from assets.ga_ultraplan import phase, parallel
-
-# --- file wiring (shared notes under a task folder) ---
-work = "temp/<task-slug>"
-ctx, attempts, criteria = f"{work}/context.md", f"{work}/attempts.md", f"{work}/criteria.md"
-
-# --- task decomposition ---
-# Three independent blades on the candidate space; each excludes recorded repeats.
-hunters = [
-    ("Known-pattern hunter", "candidates from established patterns only"),
-    ("Weird-angle hunter",   "unconventional candidates; argue why each could work"),
-    ("Constraint hunter",    "candidates that exploit the constraints / narrow success condition"),
-]
-
-with phase("Expand candidate frontier", "independent lenses, avoid recorded duplicates"):
-    notes = parallel([
-        (name, f"Read {ctx}, {attempts}, {criteria}. Propose {lens}.") for name, lens in hunters
-    ], max_workers=3)
-
-with phase("Reduce candidates", "dedupe and pick what to attempt"):
-    shortlist = parallel([
-        ("Candidate reducer", f"Read {attempts} and outputs {notes}. Emit a deduped attempt list, each with explicit success evidence.")
-    ])[0]
-
-with phase("Attempt and verify", "strongest candidates, physical proof only"):
-    results = parallel([
-        ("Attempt + verify", f"Read {shortlist}. Try the strongest candidates; report physical evidence of success or failure per candidate.")
-    ])
-
-print(results)
-```
-
-The three blades above (`known-pattern / weird-angle / constraint`) are the **most generic** placeholder. They are not the answer for every Hunt. After scouting, **re-instantiate the lenses from the domain's real structure**: the blades should encode *competing hypotheses about where this specific problem is hard*, so the reducer can kill the wrong hypothesis from evidence and concentrate effort on the live one. If your three lenses could apply unchanged to any task, you have not decomposed yet. When candidates need *attempt → verify per item*, the attempt phase is a `mapchain(shortlist, attempt_step, verify_step)`, not a flat `parallel`.
-
-### Improve template
-
-```python
-from assets.ga_ultraplan import phase, parallel
-
-# wiring: shared notes under a task folder
-work = "temp/<task-slug>"
-target, criteria, findings = f"{work}/target.md", f"{work}/criteria.md", f"{work}/findings.md"
-
-# decomposition: four non-overlapping critical angles, each hunting a distinct failure class
-critics = [
-    ("Correctness critic", "exact correctness failures, citing target fragments"),
-    ("Simplicity critic",  "over-complex or confused parts; propose a simpler structure"),
-    ("Verification critic","claims or changes that lack executable verification"),
-    ("Originality critic", "derivative, generic, or weak ideas; propose stronger framing"),
-]
-
-with phase("Sharp independent critique", "specific failure modes, not generic review"):
-    critiques = parallel([
-        (name, f"Read {target}, {criteria}, {findings}. Find {angle}.")
-        for name, angle in critics
-    ], max_workers=4)
-
-with phase("Prioritize improvements", "reject vague or duplicate criticism"):
-    issue_list = parallel([
-        ("Issue reducer", f"Read critiques: {critiques}. Prioritize issues with exact location, evidence, fix direction. Reject duplicates and vague comments.")
-    ])[0]
-
-print(issue_list)
-```
-
-Usually the next script after this template is a **Sweep** over accepted issues or target sections. For a harder draft, run the critics as **competing rewrites** (each critic emits a full alternative), then a reducer picks the strongest per section.
-
-### Explore template
-
-```python
-from assets.ga_ultraplan import phase, parallel
-
-# wiring: shared notes under a task folder
-work = "temp/<task-slug>"
-context, findings = f"{work}/context.md", f"{work}/findings.md"
-
-# decomposition: four mappers triangulating the same unknown from independent directions
-mappers = [
-    ("Surface mapper",       "enumerate visible components/files/endpoints/sources and what each proves"),
-    ("Mechanism mapper",     "how the relevant mechanism likely works; separate evidence from inference"),
-    ("Contradiction mapper", "evidence that contradicts the current assumptions"),
-    ("Frontier mapper",      "unknowns whose answers would change the next archetype"),
-]
-
-with phase("Map unknown space", "evidence first, no premature final answer"):
-    maps = parallel([
-        (name, f"Read {context}, {findings}. Investigate: {lens}.")
-        for name, lens in mappers
-    ], max_workers=4)
-
-with phase("Reduce map", "knowns, unknowns, contradictions, next archetype"):
-    frontier = parallel([
-        ("Exploration reducer", f"Read {findings} and outputs: {maps}. Write knowns, unknowns, contradictions, and recommended next archetype.")
-    ])[0]
-
-print(frontier)
-```
-
-### Sweep template
-
-```python
-from assets.ga_ultraplan import phase, mapchain, parallel
-
-# wiring: shared notes under a task folder; item list is controller-produced/validated
-work = "temp/<task-slug>"
-context, criteria, items_file = f"{work}/context.md", f"{work}/criteria.md", f"{work}/items.txt"
-items = [l.strip() for l in open(items_file, encoding="utf-8") if l.strip()]
-
-# decomposition: same two-step chain per item (inspect -> verify), every item, none skipped
-with phase("Per-item sweep", "no duplicate or omitted items"):
-    item_reports = mapchain(
-        items,
-        ("Inspect {item}", f"Read {context}, {criteria}. Inspect ONLY {{item}}. Output item ID, evidence, proposed action, unresolved risk."),
-        ("Verify {previous}", "Tail {previous}. Verify the report against tools/files. Keep only supported claims; note gaps."),
-        max_workers=6,
-    )
-
-with phase("Coverage reduce", "coverage table and omissions"):
-    coverage = parallel([
-        ("Coverage reducer", f"Read item list {items_file} and reports: {item_reports}. Produce coverage count, omissions, accepted findings, and next missing-set sweep if needed.")
-    ])[0]
-
-print(coverage)
-```
-
-If you wrote `reviews = parallel(review tasks); verified = parallel(verify tasks)` but each item only needs its own prior review, rewrite it as `mapchain`. For very large item sets, sweep in **bounded batches** (chunk the list, reduce coverage per batch) so the controller can read results between batches.
-
-## Dynamic fan-out and parsing discipline
-
-Avoid asking workers to return complex JSON task graphs. For dynamic fan-out:
-
-1. Prefer controller scouting to produce a plain item/control list before the script.
-2. Or have a planner/reducer write a plain text list file with one item per line.
-3. Validate the list with Python before using it for `mapchain`.
-4. If the list changes the workflow shape, stop and regenerate the next script instead of contorting the current script.
-
-Only parse small, explicit control data. Rich reasoning belongs in `.out.txt` files and shared markdown files.
-
-When a prompt both interpolates wiring (`f"...{ctx}..."`) and shows a worker a literal `{item}`/`{previous}` placeholder, the f-string will eat the literal braces — escape them as `{{item}}`, or the worker receives a blank.
-
-For loop-until-dry discovery, stop only after K consecutive rounds surface nothing new — a fixed `while count < N` misses the tail. Dedup each round against everything **seen**, not against what was **confirmed**: deduping on confirmed lets judge-rejected items reappear every round, so the loop never converges.
-
-## Output discipline
-
-Worker outputs should end with a compact, easy-to-tail conclusion:
-
-```text
-## Conclusion
-- Item / lens:
-- Evidence inspected:
-- Finding or attempt result:
-- Confidence / failure reason:
-- What should happen next:
-```
-
-Reducers should produce controller-ready files:
-
-```text
-## Decision state
-- Accepted facts:
-- Rejected / unproven claims:
-- Attempts or decisions recorded:
-- Remaining frontier:
-- Recommended next archetype:
-- Coverage bounds / skipped scope:
-```
-
-Downstream prompts should usually say `Tail <previous>.out.txt` rather than `Read <previous>.out.txt` to avoid re-ingesting the full transcript noise.
-
-## Final user-facing answer requirements
-
-When UltraPlan is used to produce a plan, the final answer must include:
-
-- what archetype(s) were run and what scope was covered;
-- the concrete decision or plan, not just that subagents completed;
-- ordered steps with responsible files/items when implementation is involved;
-- evidence or output files that support the key claims;
-- verification steps that can actually be executed;
-- risks, skipped scope, unresolved questions, and why they do or do not block progress;
-- whether another UltraPlan wave is recommended.
-
-Do not claim exhaustive coverage unless the item list, coverage notes, and reducer prove it. Do not pad with generic advice.
-
-## Safety and execution rules
-
-- Read relevant code, files, pages, logs, or records before planning edits. UltraPlan must be grounded in the actual target.
-- Keep irreversible operations out of worker prompts unless the user explicitly approved them.
-- Subagents have full tools and physical execution ability; give precise boundaries, allowed targets, and stop conditions.
-- Prefer file paths over copied long text in prompts. Tell workers exactly which shared files and output files to inspect.
-- After a script completes, the main agent must read the final output file and sanity-check important intermediate files.
-- Use several smaller scripts when the next phase depends on prior results; do not hide recursive control inside one huge script.
-- If a worker fails, inspect its `.out.txt` or error, update state, reduce scope or timeout, and retry with new information. Do not repeat blindly.
-- If coverage is bounded by time, sampling, top-N, timeout, excluded subsystem, or failed tool, record the bound in the relevant shared note and the final answer.
-
-## Minimal controller skeleton
-
-```python
-from assets.ga_ultraplan import phase, parallel, mapchain
-
-# wiring: shared notes under a task folder
-work = "temp/<task-slug>"
-ctx, attempts, criteria = f"{work}/context.md", f"{work}/attempts.md", f"{work}/criteria.md"
-
-with phase("Current archetype", "state what this script is trying to learn or reduce"):
-    outputs = parallel([
-        ("Lens A", f"Read {ctx}, plus {attempts} or {criteria} if relevant. Do the assigned independent work."),
-        ("Lens B", f"Read {ctx}, plus {attempts} or {criteria} if relevant. Use a different lens; avoid duplicating Lens A."),
-    ])
-
-with phase("Reduce", "controller-ready decision state"):
-    reduced = parallel([
-        ("Reducer", f"Read the relevant shared notes and these outputs: {outputs}. Write accepted facts, rejected claims, frontier, and recommended next archetype.")
-    ])[0]
-
-print(reduced)
-```
-
-After this script exits, the main agent reads `reduced`, updates shared state, and decides whether to answer, act directly, ask the user, or write the next UltraPlan script.