fix httpapp output auth and update ultraplan docs

assets/ga_httpapp.py

@@ -1,51 +1,125 @@
 import threading, sys, os
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-from fastapi import FastAPI, Header, HTTPException, Query
+from fastapi import FastAPI, Header, HTTPException, Query, Depends; from fastapi.responses import HTMLResponse
 from pydantic import BaseModel
-from agentmain import GeneraticAgent
+from agentmain import GenericAgent as GA
 
 PORT, API_KEY = int(sys.argv[1]), sys.argv[2]
-app, agent, lock = FastAPI(), GeneraticAgent(), threading.Lock()
-task_id, outputs, stopped, base = 0, [], True, 0
+app, agent, lock = FastAPI(), GA(), threading.Lock()
+outputs, stopped = [], True
 threading.Thread(target=agent.run, daemon=True).start()
-class Req(BaseModel): prompt: str
+class Req(BaseModel): prompt: str = ""
 agent.verbose = False
 
-def auth(key):
-    if key != API_KEY: raise HTTPException(404)
+def require_key(key: str = Query(None), x_api_key: str = Header(None, alias="X-API-Key")):
+    if API_KEY not in (key, x_api_key): raise HTTPException(404)
 
-def run_task(tid, prompt):
+def run_task(prompt):
     global stopped
+    segs = []  # 本任务按 turn 索引的分段输出
+    with lock: task_start = len(outputs)
+    def flush():
+        with lock: outputs[task_start:] = segs
     try:
         dq = agent.put_task(prompt, source="http")
-        while "done" not in (item := dq.get(timeout=1200)):
-            if "next" in item:
-                t = base + item.get("turn", 0)
-                while len(outputs) <= t: outputs.append({"task_id": tid, "text": ""})
-                outputs[t] = {"task_id": tid, "text": item["outputs"][-1] if item.get("outputs") else ""}
-        for i, txt in enumerate(item.get("outputs", [])):
-            t = base + i
-            while len(outputs) <= t: outputs.append({"task_id": tid, "text": ""})
-            outputs[t] = {"task_id": tid, "text": txt}
+        while "done" not in (item := dq.get(timeout=2200)):
+            outs = item.get("outputs")
+            if not outs: continue
+            idx = max(0, int(item.get("turn", 0) or 0) - 1)  # turn 1-based → 槽位 0-based
+            while len(segs) <= idx: segs.append("")
+            segs[idx] = str(outs[-1])                         # 当前 turn
+            if len(outs) >= 2 and idx >= 1: segs[idx - 1] = str(outs[-2])  # 前一 turn 落定值
+            flush()
+        segs = [str(s) for s in item.get("outputs", [])]      # done 时全量替换
+        flush()
     finally: stopped = True
 
 @app.post("/put_task")
-def put_task(req: Req, key: str = Header(alias="X-API-Key")):
-    auth(key); global task_id, stopped, base
+def put_task(req: Req, _=Depends(require_key)):
+    global stopped
     with lock:
         if not stopped: return {"ok": False, "error": "should abort first"}
-        stopped = False; task_id += 1; base = len(outputs)
-    threading.Thread(target=run_task, args=(task_id, req.prompt), daemon=True).start()
-    return {"ok": True, "task_id": task_id}
+        stopped = False
+    threading.Thread(target=run_task, args=(req.prompt,), daemon=True).start()
+    return {"ok": True}
 
 @app.post("/abort")
-def abort(key: str = Header(alias="X-API-Key")): auth(key); agent.abort(); return {"ok": True}
+def abort(_=Depends(require_key)): agent.abort(); return {"ok": True}
+
+@app.post("/input")
+def input_task(req: Req, _=Depends(require_key)):
+    global stopped
+    if not stopped: agent.intervene = req.prompt; return {"ok": True, "mode": "intervene"}
+    with lock:
+        if not stopped: agent.intervene = req.prompt; return {"ok": True, "mode": "intervene"}
+        stopped = False
+    threading.Thread(target=run_task, args=(req.prompt,), daemon=True).start()
+    return {"ok": True, "mode": "task"}
 
 @app.get("/output")
-def get_output(key: str = Header(alias="X-API-Key"), k: int = Query(5)):
-    auth(key)
-    recent = [o["text"] for o in outputs[-k:]] if outputs else []
-    return {"task_id": task_id, "stopped": stopped, "output": "\n".join(recent)}
+def get_output(k: int = Query(5), _=Depends(require_key)):
+    with lock: r = outputs[-k:]
+    return {"stopped": stopped, "output": "\n".join(r),
+            "history": "\n".join(str(h) for h in agent.history)}
+
+@app.get("/llm")
+def llm_ep(llm_no: int = Query(None), _=Depends(require_key)):
+    if llm_no is not None:
+        agent.next_llm(llm_no)
+    return {"llm_no": agent.llm_no, "name": agent.get_llm_name(),
+            "llms": [{"no": i, "name": n, "current": a} for i, n, a in agent.list_llms()]}
+
+@app.get("/sysprompt")
+def sysprompt_ep(text: str = Query(None), _=Depends(require_key)):
+    if text is not None:
+        agent.extra_sys_prompts = [text] if text else []
+    return {"extra_sys_prompts": agent.extra_sys_prompts}
+
+HELP = """GA HTTP 操作协议（所有请求带 ?key=API_KEY，或 Header X-API-Key）
+GET  /output?k=N      查看状态：{stopped, output(末N条), history}。stopped=true 表示空闲
+POST /input  {prompt} 下发指令：空闲时作为新任务，忙时作为中途干预(intervene)
+POST /abort           中止当前任务
+GET  /llm[?llm_no=N]  查/切模型：返回 {llm_no,name,llms:[{no,name,current}]}
+GET  /sysprompt[?text]查/设附加系统提示(extra_sys_prompts)，text 为空则清空
+纠偏流程：先 GET /output 读 history 判断状态→需要时 POST /input 注入纠偏指令"""
+
+@app.get("/help")
+def help_ep(_=Depends(require_key)): return {"help": HELP}
+
+@app.get("/")
+def ui():
+    return HTMLResponse(f"""<!DOCTYPE html>
+<html><head><meta charset="UTF-8"><meta name="viewport" content="width=device-width,initial-scale=1">
+<title>GA Monitor</title>
+<script src="https://cdn.jsdelivr.net/npm/marked/marked.min.js"></script>
+<style>
+*{{margin:0;padding:0;box-sizing:border-box}}
+body{{font:14px/1.6 system-ui,sans-serif;background:#f8f9fa;color:#212529;padding:24px;max-width:900px;margin:0 auto}}
+#status{{font-size:18px;padding:8px 16px;border-radius:8px;margin-bottom:16px;display:inline-block;font-weight:600}}
+.stopped{{background:#d4edda;color:#155724}}.running{{background:#fff3cd;color:#856404}}
+.section{{background:#fff;border-radius:10px;padding:18px;margin-bottom:14px;box-shadow:0 1px 4px rgba(0,0,0,.06)}}
+.section h3{{color:#533483;margin-bottom:8px;font-size:15px}}
+textarea{{width:100%;height:80px;background:#fff;color:#212529;border:1px solid #ced4da;border-radius:6px;padding:10px;font:inherit;resize:vertical}}
+button{{padding:10px 24px;background:#533483;color:#fff;border:none;border-radius:6px;font:inherit;cursor:pointer;font-weight:500}}
+button:hover{{background:#7c3aed}}
+pre{{background:#f1f3f5;padding:12px;border-radius:6px;overflow-x:auto;font:13px monospace;max-height:400px;overflow-y:auto}}
+</style></head><body>
+<div id="status" class="stopped">● Loading...</div>
+<div class="section"><h3>Output</h3><div id="output"></div></div>
+<div class="section"><h3>History</h3><div id="history"></div></div>
+<textarea id="prompt" placeholder="Enter instruction..."></textarea>
+<button onclick="send()">Send</button>
+<script>
+const K=new URLSearchParams(location.search).get('key')||'';
+async function poll(){{let r=await fetch('/output',{{headers:{{'X-API-Key':K}}}});let d=await r.json();
+let s=document.getElementById('status');s.textContent=(d.stopped?'● Stopped':'● Running');s.className=d.stopped?'stopped':'running';
+document.getElementById('output').innerHTML=marked.parse(d.output||'_empty_');
+document.getElementById('history').innerHTML=marked.parse(d.history||'_empty_');}}
+async function send(){{let p=document.getElementById('prompt').value;if(!p)return;
+await fetch('/input',{{method:'POST',headers:{{'X-API-Key':K,'Content-Type':'application/json'}},body:JSON.stringify({{prompt:p}})}});
+document.getElementById('prompt').value='';poll();}}
+poll();setInterval(poll,3000);
+</script></body></html>""")
 
 if __name__ == "__main__":
     import uvicorn; uvicorn.run(app, host="0.0.0.0", port=PORT)

assets/ga_ultraplan.py

@@ -6,7 +6,9 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
 __all__ = ["phase", "parallel", "mapchain"]
 
-_T0 = time(); _RUN_DIR = None; _phases = []; _phase_stack = []; _tasks = []; _current = "idle"; _events = []; _srv = None; _last = time(); _lock = threading.Lock()
+_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()}")
+os.makedirs(_RUN_DIR, exist_ok=True)
 
 def _note(s):
     global _last
@@ -91,11 +93,7 @@ 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 _RUN_DIR
     root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
-    os.makedirs(tmp := os.path.join(root, "temp"), exist_ok=True)
-    if _RUN_DIR is None:
-        _RUN_DIR = os.path.join(tmp, f"ultraplan_{int(_T0)}_{os.getpid()}"); os.makedirs(_RUN_DIR, exist_ok=True)
     with tempfile.NamedTemporaryFile("w", encoding="utf-8", suffix=".txt", prefix="ultra_", dir=_RUN_DIR, delete=False) as f:
         f.write(desc if prompt is None else prompt)
     print(f"[subagent] {desc} -> {f.name}", flush=True); _note(f"agent: {desc}")

llmcore.py

@@ -59,6 +59,7 @@ def compress_history_tags(messages, keep_recent=10, max_len=800, force=False, in
                 if not isinstance(b, dict): continue
                 t = b.get('type')
                 if t == 'text' and isinstance(b.get('text'), str): b['text'] = _trunc(b['text'])
+                elif t == 'thinking' and isinstance(b.get('thinking'), str): b['thinking'] = _trunc_str(b['thinking'])
                 elif t == 'tool_result':
                     tc = b.get('content')
                     if isinstance(tc, str): b['content'] = _trunc_str(tc)

memory/ultraplan_sop.md

@@ -16,10 +16,11 @@ main agent controller
 
 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; shared state lives in explicit files under the current working directory, usually `./temp/ultraplan_<task>/`.
+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
 
@@ -31,10 +32,12 @@ from assets.ga_ultraplan import phase, parallel, mapchain
 
 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.
 
+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.
+
 Each task can be one of:
 
 ```python
-("Short display name", "Prompt template with {item}, {previous}, {state_dir}, ...")
+("Short display name", "Prompt template with {item}, {previous}, ...")
 
 {
     "desc": "Short display name with {item}",
@@ -51,9 +54,9 @@ A task returns the path of its subagent output file, normally `.../temp/ultra_xx
 
 ## 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, create only the shared files this task needs, under the current working directory, usually `./temp/ultraplan_<target>/`.
+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. Use task-shaped files such as `sop_context.md`, `current_findings.md`, `candidate_urls.txt`, `failed_attempts.md`, `coverage_list.txt`, or `acceptance_criteria.md`. 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.
+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:
 
@@ -62,6 +65,8 @@ Rules:
 - 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
 
@@ -98,12 +103,10 @@ Each step receives the current value as `{item}` and `{previous}`. Initially thi
 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
-context_file = f"{state_dir}/context.md"
 reports = mapchain(
     items,
-    ("Inspect {item}", f"Read {context_file} and inspect assigned item: {{item}}. Output evidence-backed findings."),
+    ("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."),
-    state_dir=state_dir,
 )
 ```
 
@@ -125,6 +128,8 @@ Use this cycle:
 
 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.
@@ -221,88 +226,106 @@ Do not force all transformations into one script. If the next archetype depends
 
 ## Script templates by archetype
 
-The examples below are starting points. Replace placeholders with concrete shared files and target objects. Keep scripts short enough that the controller can read outputs and decide the next 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
 
-```python
-# Run from GA code root.
-from assets.ga_ultraplan import phase, parallel, mapchain
+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.
 
-state_dir = "./temp/ultraplan_<target>"
-context_file = f"{state_dir}/context.md"      # task-shaped target/context note
-attempts_file = f"{state_dir}/attempts.md"    # tried candidates and rejected repeats
-criteria_file = f"{state_dir}/criteria.md"    # success evidence or constraints, if useful
+```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"):
-    candidate_notes = parallel([
-        ("Known-pattern hunter", f"Read {context_file} and {attempts_file}. Propose candidates from known patterns only; exclude repeats."),
-        ("Weird-angle hunter", f"Read {context_file} and {attempts_file}. Propose unconventional candidates; explain why each might work."),
-        ("Constraint hunter", f"Read {context_file} and {criteria_file}. Propose candidates that exploit constraints or narrow success criteria."),
-    ], state_dir=state_dir, max_workers=3)
+    notes = parallel([
+        (name, f"Read {ctx}, {attempts}, {criteria}. Propose {lens}.") for name, lens in hunters
+    ], max_workers=3)
 
-with phase("Reduce candidates", "dedupe and select attempts"):
+with phase("Reduce candidates", "dedupe and pick what to attempt"):
     shortlist = parallel([
-        ("Candidate reducer", f"Read {attempts_file} and these outputs: {candidate_notes}. Write a deduplicated attempt list with success evidence for each.")
+        ("Candidate reducer", f"Read {attempts} and outputs {notes}. Emit a deduped attempt list, each with explicit success evidence.")
     ])[0]
 
-with phase("Attempt and verify", "each candidate advances independently"):
-    # In real use, the controller may parse/validate shortlist into candidate IDs before this phase.
+with phase("Attempt and verify", "strongest candidates, physical proof only"):
     results = parallel([
-        ("Attempt selected candidates", f"Read {shortlist}. Try the strongest candidates. For each, report physical evidence of success or failure.")
+        ("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
 
-state_dir = "./temp/ultraplan_<target>"
-target_file = f"{state_dir}/target_or_draft.md"
-criteria_file = f"{state_dir}/criteria.md"          # optional constraints/current requirements
-findings_file = f"{state_dir}/current_findings.md"  # optional known claims or prior verification notes
+# 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([
-        ("Correctness critic", f"Read {target_file} and {criteria_file}. Find exact correctness failures; cite target fragments."),
-        ("Simplicity critic", f"Read {target_file}. Find over-complex or confused parts; propose simpler structure."),
-        ("Verification critic", f"Read {target_file} and {findings_file}. Find claims or changes that lack executable verification."),
-        ("Originality critic", f"Read {target_file}. Identify derivative, generic, or weak ideas; propose stronger framing."),
-    ], state_dir=state_dir, max_workers=4)
+        (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}. Produce prioritized issues with exact target location, evidence, and fix direction. Reject duplicates and vague comments.")
+        ("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.
+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
 
-state_dir = "./temp/ultraplan_<target>"
-context_file = f"{state_dir}/context.md"          # what is known about the unknown space
-findings_file = f"{state_dir}/current_findings.md" # optional assumptions or prior evidence
+# 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([
-        ("Surface mapper", f"Read {context_file}. Enumerate visible components/files/endpoints/sources and what each proves."),
-        ("Mechanism mapper", f"Read {context_file}. Investigate how the relevant mechanism likely works; separate evidence from inference."),
-        ("Contradiction mapper", f"Read {context_file} and {findings_file}. Search for evidence that contradicts the current assumptions."),
-        ("Frontier mapper", f"Read {context_file}. List unknowns whose answers would change the next archetype."),
-    ], state_dir=state_dir, max_workers=4)
+        (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_file} and outputs: {maps}. Write knowns, unknowns, contradictions, and recommended next archetype.")
+        ("Exploration reducer", f"Read {findings} and outputs: {maps}. Write knowns, unknowns, contradictions, and recommended next archetype.")
     ])[0]
 
 print(frontier)
@@ -313,18 +336,17 @@ print(frontier)
 ```python
 from assets.ga_ultraplan import phase, mapchain, parallel
 
-state_dir = "./temp/ultraplan_<target>"
-context_file = f"{state_dir}/context.md"       # target-specific context or source references
-criteria_file = f"{state_dir}/criteria.md"     # optional constraints/checklist
-items_file = f"{state_dir}/items.txt"          # one item per line, produced or checked by the controller
-items = [line.strip() for line in open(items_file, encoding="utf-8") if line.strip()]
+# 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_file} and {criteria_file}. Inspect assigned item only: {{item}}. Output item ID, evidence, proposed action, unresolved risk."),
-        ("Verify {previous}", "Tail {previous}. Verify the item report against tools/files. Keep only supported claims and note gaps."),
-        state_dir=state_dir,
+        ("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,
     )
 
@@ -336,7 +358,7 @@ with phase("Coverage reduce", "coverage table and omissions"):
 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`.
+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
 
@@ -349,6 +371,10 @@ Avoid asking workers to return complex JSON task graphs. For dynamic fan-out:
 
 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:
@@ -406,16 +432,15 @@ Do not claim exhaustive coverage unless the item list, coverage notes, and reduc
 ```python
 from assets.ga_ultraplan import phase, parallel, mapchain
 
-state_dir = "./temp/ultraplan_<target>"
-context_file = f"{state_dir}/context.md"          # task-shaped shared context, not a required schema
-attempts_file = f"{state_dir}/attempts.md"        # optional prior attempts or coverage notes
-criteria_file = f"{state_dir}/criteria.md"        # optional success criteria or constraints
+# 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 {context_file}, plus {attempts_file} or {criteria_file} if relevant. Do the assigned independent work."),
-        ("Lens B", f"Read {context_file}, plus {attempts_file} or {criteria_file} if relevant. Use a different lens; avoid duplicating Lens A."),
-    ], state_dir=state_dir)
+        ("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([