Designing bench_env Tasks

Overview

Rushing from "here's an App" to "here are 5 task classes" produces low-difficulty suites whose judge logic can't actually verify completion. Design must precede code.

Authoritative reference: bench_env/docs/task/TASK_AUTHORING_GUIDE.md (reading §1 + §2 once is required; this skill enforces its gates).

The Gate: two artifacts before any Python

Produce both as plain text in the conversation before writing any task class. If you catch yourself opening tasks.py / defs/<TaskName>.py, stop and produce them.

1. Functional audit table (TASK_AUTHORING_GUIDE §1.1–1.2)

A table with one row per distinct feature area. Columns:

| Page/feature | Source file(s) | User-visible actions | Observable state path |

You must actually read: manifest.ts, navigation.declaration.ts, data/defaults.json, state.ts, pages/*, and the suite's app.py accessor if it exists. No skipping "because the app looks simple."

2. Data sufficiency check (TASK_AUTHORING_GUIDE §1.3)

For every function you plan to parameterize, confirm defaults.json / state.ts provides ≥3 varied entries. If it doesn't, either propose expanding defaults, or drop parameterization for that function.

Per-task: 4 judge-predict questions

For each proposed task, answer in 1-2 lines each before writing code (this is the soundness/completeness audit later enforced by TASK_AUTHORING_GUIDE §2.7 "Reliability requirements"):

1. Agent 完全做对时，最终 state / answer 长什么样？
2. Agent 最常见的 1-2 种错误是什么？会不会被误判通过？（soundness）
3. 有没有合理完成任务的替代路径？会不会被误判失败？（completeness）
4. 有无边界情况导致正确答案不唯一、或判定证据不足？

If any answer surfaces a flaw (common: initial state already equals criteria; ground truth not unique; answer requires subjective judgement), iterate the design in text — do not defer the fix to code review.

Rationalization table — STOP and do the step

Checklist before producing any class FooTask(...)

• Functional audit table in conversation
• Data sufficiency assessed per parameterized function; gaps declared
• For each task, 4 pre-sim questions answered
• No task has initial state == goal state (the task must require Agent action to flip something)
• Cross-checked TASK_AUTHORING_GUIDE.md §2.7 "Reliability requirements" + §4.7 "Authoring check_goals"

After design is approved

For actual code discipline: see the writing-bench-task-judge skill and bench_env/docs/task/TASK_CODE_SPEC.md. For tests: testing-bench-task skill + bench_env/docs/task/TASK_TESTING_GUIDE.md.

Testing bench_env Task Judges

Overview

A single positive + a single "opposite of positive" negative is not a test — it's a tautology. Real coverage means the negative exercises a failure mode an Agent could realistically produce. Most defects ship because tests stop at one reverse pattern.

Authoritative ref: bench_env/docs/task/TASK_TESTING_GUIDE.md §4 (read §4.3.2 through §4.3.5 before writing cases).

Rule 1 — Positive answer must be natural language (§4.3.2)

AnswerTask positives with bare ground-truth values (answer="32", answer="多云") bypass match_value's fuzzy/number-extraction logic — the test proves nothing.

# ❌ bare ground truth — test passes tautologically
_make_input(state, state, answer="多云")

# ✅ natural language, as an Agent would actually respond
_make_input(state, state, answer="上海今天天气多云转晴")

Format must include ground truth as substring / extractable number, plus realistic surrounding context.

Rule 2 — Pick negative patterns from the taxonomy (§4.3.3)

Blanket answer="错误答案" / curr_state == init_state is a single degenerate pattern. For each task, pick ≥1 pattern from the matching table:

Query tasks

Operate tasks

Cross-app tasks

Every negative case must name a pattern from the relevant table. Complex tasks (multi-field, cross-app) need ≥2 different patterns.

Rule 3 — hybrid tasks need ≥2 negatives (§4.3.1)

hybrid 任务同时检查状态变更和 answer，失败维度独立，≥2 反例：

Rule 4 — Cover match_value boundary cases at suite level (§4.3.4)

These are suite-level concerns; add them as extra cases (name them "TaskName_suffix" and put them in HYBRID_EXTRA_NEGATIVE_CASES or similar to avoid breaking test_*_judge_matrix_complete).

Pick representative tasks and cover:

• 中文数字: at least one positive with answer="北京现在二十度"
• 空 answer: at least one AnswerTask negative with answer=None
• 多数字干扰: GT value in answer while other numbers are also present
• 子串包含陷阱: for any 肯定/否定 query, negative proves "通过" ⊂ "未通过" handled
• 尾零格式: for 小数金额 tasks, positive with "总共278.20元"

Rule 5 — Multi-format positives for structured values (§4.3.5)

When get_answer() returns time ("09:54") or duration ("0小时59分"), match_value substring fails on equivalent formats. If the task uses match_time / match_duration in check_goals, the test must parametrize positives:

@pytest.mark.parametrize("answer", [
    "G7010，1小时10分，13:10",         # exact
    "G7010，70分钟，下午1点10分",       # chinese natural
    "G7010，1小时10分钟，13:10到达",    # mixed format
], ids=["exact", "chinese_natural", "mixed"])
async def test_fastest_train_flexible(self, env, answer): ...

Rule 6 — Keep matrix-complete invariant intact

test_*_judge_matrix_complete requires exactly one main positive + one main negative per task name. Extra variants go into a separate list (e.g. HYBRID_EXTRA_NEGATIVE_CASES) with suffixed names ("CheckBalance_empty_answer").

Rationalization table

Checklist

• Each task has ≥1 positive with natural-language answer
• Each task has ≥1 negative; pattern named from §4.3.3 tables (not mirror-of-positive)
• hybrid tasks have ≥2 negatives covering each independent fail axis
• Suite-level: §4.3.4 boundary cases sampled (中文数字 / 空 answer / 多数字干扰 / 子串陷阱 / 尾零)
• Structured-value tasks (time / duration / date): §4.3.5 parametrized positives
• New cases don't break test_*_judge_matrix_complete (variants go to *_EXTRA_*)
• Uses make_judge_input from conftest.py; no defensive .get() on state
• Ran pytest bench_env/tests/<suite>/ -m "not live" -v and it passes

Writing bench_env Task Judges

Overview

A judge decides whether the Agent completed the task. Two failure modes dominate:

• Soundness hole — an obviously wrong path is judged passed=True (keyword hit, unstable trace field).
• Completeness hole — a reasonable completion path is judged passed=False (bound to specific wording or UI step).

Neither is caught by type checks — both are caught by applying the CRUD model rigorously.

Authoritative refs:

• bench_env/docs/task/TASK_AUTHORING_GUIDE.md — task design + CRUD judge model (read §2 in full)
• bench_env/docs/task/TASK_CODE_SPEC.md — code rules (file responsibilities, naming, defensive-coding ban, time APIs, CriteriaTask)
• bench_env/docs/REFERENCE.md — JudgeInput / JudgeResult field & expected_changes path syntax lookup
• bench_env/docs/task/GROUNDED_MODE.md — only if task uses answer_fields / grounded eval

Step 1 — Classify the task as CRUD

Before writing any check, name the operation out loud:

Each CRUD type has ONE correct strategy — not a choice. Using lookup-on-current for Delete/Create creates false positives (sampler bug → passes as Agent action).

Step 2 — Survey the App module for reusable abstractions

Before writing any judge, Read app.py end-to-end with the Read tool (split into sections only if the file exceeds the tool's limit). Scan every def line plus its docstring / leading comment, plus any module-level helper functions above the class. Do not substitute grep for this. Grep silently misses:

• module-level helpers (_note_text, _pick_keywords_from_note) that sit above the class
• @staticmethod / @property / @classmethod methods that your prefix guess won't anticipate
• helpers whose names don't follow the shape you grepped for (latest_note_by_title is a find-by-name primitive but doesn't start with find_; visible_notes is a filtered view but doesn't start with filtered_)
• _xxx private methods that contain logic you could extract

Reading the file end-to-end is cheap (one tool call for a few hundred lines). Missing a helper and duplicating it is expensive — the duplicate anchors every future author to the wrong template, and the trap rationale that belonged on the App method ends up lost. After app.py, skim tasks.py as well: existing tasks show you the canonical call shape to mirror.

As you read, classify each method against TASK_AUTHORING_GUIDE.md §2.1 "Three-tier App helpers" (that section is the backbone; re-read it if rusty). The table below is a fast-lookup of the layers a judge most often needs — use it as a mental checklist while scanning, not as a grep target list:

Decision once you've surveyed:

1. An existing check_* covers your goal → call it from check_goals.
2. A lower-layer primitive exists (e.g. removed_X_ids()) but no check_* wraps it → add a new check_* method in the same file, following the surrounding naming convention (check_<noun>_<verb-past> or check_<noun>_<adjective>), call it from check_goals.
3. Nothing exists → add both the primitive and the check_* to app.py, then call the check_* from check_goals.

Don't inline set-ops, id diffs, or find-by loops in check_goals. Doing so:

• duplicates logic that already exists or will exist
• hides the reason the check is correct (e.g. "why id-diff, not text lookup" — the trap rationale lives in the App method's docstring, not in every task that uses it)
• anchors the next author to a bad template, because they'll copy the most recent task

Concrete counter-example (what NOT to do):

# ❌ inline logic in check_goals — duplicates what should be an App method
def check_goals(self, input):
    notes = Notes(input.apps["notes"], init=input.apps_init["notes"])
    init_target = notes.init.find_todo_by_text(self.p.todo_text)
    target_id = str((init_target or {}).get("id") or "")
    current_ids = {str(t.get("id") or "") for t in notes.todos}
    removed = bool(target_id) and target_id not in current_ids
    return [{"field": "todo_deleted", "expected": self.p.todo_text,
             "actual": None if removed else self.p.todo_text, "passed": removed}]

# ✅ push to Notes.check_todo_deleted(text), call it here
def check_goals(self, input):
    notes = Notes(input.apps["notes"], init=input.apps_init["notes"])
    return [notes.check_todo_deleted(self.p.todo_text)]

Step 3 — Enforce the init/current rule

# ✅ Modify: init resolves identity, current verifies result
init_alarm = self.init.find_alarm_by_time(h, m)
assert init_alarm is not None  # sampler contract
alarm = self.find_alarm_by_id(init_alarm["id"])  # current

Step 4 — Sampler contract asserts (only where upstream is silent)

Add assert in an App check_* method only when the target's presence in init isn't already enforced upstream and the natural failure mode would be silent passed=False. The purpose is attribution: route sampler bugs to judge_error instead of passed=False. If upstream already raises, a check_*-level assert re-validates the same premise and violates TASK_CODE_SPEC.md §5 "No defensive coding".

Upstream is already enforced (skip the assert — it's defensive coding):

• Sampler raises on no candidate — the canonical pattern: _sample_X reads env_state, picks a candidate, raise ValueError if none. By the time check_* runs, the target is guaranteed to be in init. Re-asserting is double-validation.
• Data method raises on not-found (find_contact_wxid(name) → ValueError) — premise already enforced by the lookup itself.
• Query / direct leaf access — natural dereference (alarm["note"], self.get("settings.darkMode")) raises KeyError/TypeError → already judge_error.
• target is None is a legitimate Agent failure mode — e.g. Agent deleted the alarm it was supposed to modify → alarm is None → passed=False is correct per TASK_AUTHORING_GUIDE.md §2.5 "Modify".

Add assert when upstream is silent or absent (sampler bug would be silently absorbed into passed=False):

• Legacy silent-fallback sampler — older sampler that returns a default instead of raising when no candidate.
• Hardcoded default param with no sampler: binding — CLI overrides, smoke tests, suites that don't sample this field. No upstream guarantee exists.
• Externally-injected params — cross-app contracts, harness-level inputs, params whose provenance lives outside the suite.

# ✅ Sampler raises on no candidate → no assert; target is guaranteed in init
# (e.g. Notes._sample_incomplete_todo raises ValueError if no candidate)
def check_todo_deleted(self, text):
    init_todo = self.init.find_todo_by_text(text)  # guaranteed non-None by sampler
    target_id = str(init_todo["id"])
    return {...}

# ✅ No sampler binding for `alarm_id` → param may come from CLI/default → assert
def check_deleted_alarm(self, alarm_id):
    assert self.init.find_alarm_by_id(alarm_id) is not None, (
        f"Upstream bug: alarm {alarm_id} not in init"
    )
    removed = self.removed_alarm_ids()
    return {"field": "alarm_deleted", "expected": alarm_id,
            "actual": sorted(removed), "passed": str(alarm_id) in removed}

Quick decision: trace where the target id/text came from. If the path is env_state → _sample_X (raises on empty) → params.X → check_*, the assert is redundant. If the path is default = "..." → params.X → check_* or anything involving user-supplied / hardcoded values, keep the assert.

Asserts live in App check_* methods, not in tasks.py's check_goals().

Step 5 — Declarative first, check_goals last (TASK_AUTHORING_GUIDE §4.2)

Decision order when writing a task:

1. Can it be answer = ".path" / answer = ".path[field={param}].x"? → use it.
2. Can it be criteria = {"key": "{param}"}? → use it.
3. Only if neither fits, write get_answer() / check_goals().

# ❌ Over-written
def get_answer(self, input):
    for p in Railway12306(input.apps["railway12306"]).passengers:
        if p["isDefault"]:
            return p["name"]

# ✅ One line
answer = ".passengers[isDefault=True].name"

Step 6 — Soundness + completeness self-audit (TASK_AUTHORING_GUIDE §2.7 "Reliability requirements")

Before committing, walk through:

If any answer is "yes," rework the check — either strengthen evidence (for soundness) or broaden acceptance (for completeness).

Side-effect fence: precise expected_changes first, conservation check second

expected_changes (see bench_env/docs/REFERENCE.md "expected_changes path syntax") is the framework's automatic diff gate — any path outside it fails the task regardless of check_goals. Narrow it before you reach for a hand-written conservation check:

• CriteriaTask auto-derives expected_changes from criteria keys — usually no declaration needed.
• Precise paths let the framework do the whole conservation job:
  • expected_changes = ["todos[id={todo_id}]"] — only this todo may change; Agent deleting a different todo fails automatically.
  • expected_changes = ["moments[+1]"] — exactly one new moment may appear; adding two fails automatically.
  • expected_changes = ["contacts[name={contact}].isBlacklisted"] — only this contact's blacklist flag may change; editing their name fails automatically.
• Only when expected_changes genuinely can't be narrowed do you need a conservation check_*:
  • Batch deletes across a whole partition (DeleteAllCompletedTodos must delete an unknown-at-design-time set of ids) — pair check_X_all_deleted with check_other_partition_preserved.
  • The target id isn't exposed as a Task param, only its text/name is — precise path [id={id}] not available.
  • Multi-path Create/Modify where declaring every touched path would be brittle.

Rule of thumb: if you're about to add check_other_X_preserved(exclude_text=self.p.x) to a single-target Delete, first check whether sampler exposes the target id and whether expected_changes = ["xs[id={x_id}]"] would eliminate the check entirely. Framework-level fences beat hand-written ones — they're declarative, typechecked against the diff engine, and can't be forgotten by the next author.

Step 7 — check_goals return format (TASK_AUTHORING_GUIDE §2.4 "Check methods")

Every check dict must have exactly field / expected / actual / passed. Framework raises ValueError on missing passed.

# ❌ wrong shape
return [{"name": "route", "ok": True, "detail": {...}}]

# ✅ canonical shape, with diagnostic expected/actual
return [{"field": "newPendingOrder",
         "expected": "上海→南京 2026-03-21 G7002 二等 ×1 (赵宇轩)",
         "actual": "未创建新订单",
         "passed": False}]

One check = one semantic goal. Don't split "buy a correct ticket" into 5 field-level checks — that inflates progress and loses global diagnostic value. Do use multiple checks when there are independent goals.

expected/actual must be human-readable. expected=True, actual=None tells nobody anything.

Step 8 — Use input.apps / input.apps_init directly (TASK_CODE_SPEC §5 "No defensive coding")

# ✅
wechat = Wechat(input.apps["wechat"], init=input.apps_init["wechat"])

# ❌ defensive — masks config bugs
wechat = Wechat(input.apps.get("wechat", {}), init=(input.apps_init or {}).get("wechat"))

No or {}, no .get("xxx", {}), no try/except on data access. Data missing → exception → judge_error.

Red-flag self-check (scan before submitting)

• Named the CRUD type; picked the required strategy
• Read the suite's app.py end-to-end with the Read tool (not grepped — grep misses module-level helpers, decorated methods, and anything whose name doesn't match the prefix you guessed); classified each helper into its layer (find / diff / filtered-view / check / answer) and reused or extended existing ones instead of inlining set-ops / diffs in check_goals. Also skimmed tasks.py for the canonical call shape.
• init vs current used per the rule table
• Sampler contract assert present in Create/Delete/Modify check_* methods only when upstream is silent (legacy fallback sampler, hardcoded default without sampler:, external input); skipped when upstream already raises (canonical _sample_X pattern) to avoid violating TASK_CODE_SPEC §5 (no defensive coding); never added blindly to query / leaf-access checks
• Tried answer= / criteria= first; only wrote check_goals if declarative couldn't express it
• Soundness audit: no broad-keyword pass, no unstable trace field, no path-specific step
• Completeness audit: no forced wording / title / format
• Side-effect fence: expected_changes narrowed to precise paths ([id={x}] / [+N] / [field=v].sub) where possible; conservation check_* added only when the framework fence genuinely can't express the scope
• Check dicts use field/expected/actual/passed; expected/actual human-readable
• One check per semantic goal
• No defensive or {} / .get() on input.apps*
• Time/duration fields use match_time / match_duration (see GROUNDED_MODE.md §4.3 "Matchers"), not match_value
• Date answers use date_match_labels(date, input.os) with os_state for relative labels
• If using grounded eval: read GROUNDED_MODE.md §3-4

Rationalization table