diff --git a/docs/bugs/open/2026-05-29-code-intrinsic-hostility.md b/docs/bugs/open/2026-05-29-code-intrinsic-hostility.md new file mode 100644 index 000000000..542fe3723 --- /dev/null +++ b/docs/bugs/open/2026-05-29-code-intrinsic-hostility.md @@ -0,0 +1,281 @@ +--- +status: open # open | fixed +opened: 2026-05-29 +fixed_pr: null +priority: P1 # see per-issue priorities below +invariant_violated: docs/architecture/04_persistence.md # WAL/projection sync; doc being rewritten +related_rfc: null +--- + +# Bug: code-intrinsic hostility (footguns independent of stale docs) + +This is an umbrella findings + remediation doc. It captures places where the +**code itself** fights a contributor — silent-correctness footguns, leaks, and +fragile assumptions — separate from the (separately tracked) docs/code drift +from the sample-centric refactor. Each issue below is classified by **scope** +so we know whether the fix is a one-line method change, a schema change, a +refactor/deletion, or a design decision. + +When any single issue here grows its own repro + fix PR, split it into a +dedicated `docs/bugs/open/` entry and link back here. + +Scope legend: + +- **method fix** — change the body of one or a few methods; no schema change. +- **schema change** — Alembic migration (add/drop/alter columns or tables). +- **refactor/deletion** — move/merge/delete modules; structural. +- **design decision** — a contract question to settle before coding. +- **tests** — add invariant/roundtrip tests so the bug cannot return. + +--- + +## 1. Persistence: `refine_task` edit is silently dropped on rerun — P1 + +### Symptom + +Edit a non-running task's description via `refine_task`, then `restart_task`. +The worker executes with the **old** description. The edit appears to "take" +(the dashboard projection column updates) but has no effect on execution and +never reaches the WAL / RL replay. + +### Root cause + +The projection row stores the task **twice**: a denormalized +`SampleGraphNode.description` column *and* the full authored snapshot in +`SampleGraphNode.task_json` (`persistence/graph/models.py:45,47-55`). + +`RuntimeGraphRepository.update_node_field` updates only the column and +appends **no WAL event** (`graph_repository.py:250-277`) — unlike every sibling +mutator (`update_node_status`, `add_edge`, `update_edge_status`) which all append +via `SampleRuntimeEventAppender`. + +But execution rebuilds the Task from the **snapshot blob**, not the column: +`task_execution.py:161` passes `node.task_json` into `Task.from_definition(...)`. +So the refined description in the column is never read at rerun, and +`reconstruct_sample_runtime_state_at` (`samples/state.py:257`) never sees the +change because the WAL's `task_snapshot_json` is frozen at materialization +(`materialization.py:66,79`). + +The `refine_task` docstring claims "The graph node's description is the single +source of truth -- no definition row to keep in sync" +(`task_management.py:316-317`). That is the bug: there is no definition row, but +there *are* two in-row copies plus a WAL, and they diverge. + +### Repro + +`refine_task` → `restart_task` on the same node; assert the worker prompt / +`task_json["description"]` reflects the new value. No current test covers this +(grep `update_node_field` finds only call sites, no roundtrip assertion). + +### Scope — answering "method, schema, or refactor?" + +**Not a schema deletion.** The dual representation is intentional CQRS: + +- `sample_graph_nodes` / `sample_graph_edges` = mutable **read model** for fast + scheduling reads (`persistence/graph/models.py`). +- `sample_*_events` (7 tables) = append-only **WAL** for time-travel replay / + RL / audit (`persistence/samples/models.py`, replayed by `samples/state.py`). + +These serve different masters and both stay. The legacy `run_graph_*` and +definition tables were **already dropped** in migration +`00000003_delete_definition_runtime_columns.py`, so there is no dead table to +remove here. + +### The model: WAL is the source of truth, projection is a derived view + +The right fix is **event sourcing done properly**: the current state of a task +(and edge) is the **fold of its WAL events**, not an independently-written +projection row. Denormalized columns can never support *arbitrary* attributes a +user attaches to a task — you cannot add a column per unknown key — and writing +them in place destroys the timeline. Reconstructing from the WAL solves both: +any attribute can be mutated as an appended event, and replay-to-timestamp stays +exact. + +The codebase is already ~80% there: + +- The fold already exists — `SampleRuntimeState.from_events` / + `reconstruct_sample_runtime_state_at` (`samples/state.py:88,257`). It just + isn't the authoritative read path yet. +- A generic, timeline-preserving key/value store already exists — + `SampleAnnotationEventRow` keyed `(target_type, target_id, key)` with + `annotation.set/updated/deleted` kinds, already folded into an `annotations` + dict (`state.py:222-232`). Tasks/edges can be annotation targets today. +- `event_type` is a **free-form indexed `str`** (`samples/models.py:33,49`), not + an enum column — so new event kinds need **no migration**. + +Current state = **immutable authored base** (`task.added` snapshot: worker, +sandbox, evaluators, dependencies) ⊕ a small set of **typed editable fields** +(`description`, `assigned_worker_slug`) ⊕ an **`attributes` map under a reserved +namespace** for arbitrary user KV. The authored base never changes, so you can +always tell authored-vs-mutated and replay any point in time. The projection's +`task_json` becomes the **cached fold output** — updated only by applying the +event, never written directly — so scheduling reads stay O(1) while the WAL +remains authoritative. + +**Safety constraint:** do NOT blind-merge arbitrary user keys onto `task_json`. +A user setting an attribute named `worker` or `evaluators` must not corrupt +execution. Arbitrary KV lives in the reserved `attributes` namespace; only the +typed editable fields may override structural snapshot fields. + +### Scope — answering "method, schema, or refactor?" + +Still **not a schema change** (event kinds are free-form strings; reuse existing +tables). In order: + +1. **design decision** — settle two sub-questions: + - **Where do arbitrary attributes live?** (a) reuse `sample_annotation_events` + with `target_type="task"|"edge"` (cheapest — fold path already implemented), + or (b) add `task.attribute_changed` / `edge.attribute_changed` kinds for + clean semantic separation from out-of-band annotations. Lean (b) if + attributes are conceptually *task fields*, (a) if they're metadata/labels. + - **Edges:** do edges need arbitrary attributes, or just status (+ a label)? + Keep the pattern symmetric but don't build the full edge attribute surface + without a real use case. +2. **refactor (small, central)** — route every projection write through one + helper that appends the WAL event *and* applies it to the projection in the + same transaction, so "projection without WAL" (issue 1) and "projection + diverges from replay" (issue 4) become inexpressible. Make the editable-field + set + `attributes` overlay the fold contract in `state.py`. +3. **method fix** — `update_node_field` appends an editable-field event (and, for + arbitrary keys, an attribute/annotation event) via the helper above; point + `task_execution.py:161` at the fold output (cached `task_json`) so refinements + actually reach the worker. +4. **tests** — see issue 4 (roundtrip + mutator-completeness). + +**Effort:** still no migration. It's one design decision + a small central +refactor (the single write helper + fold contract) + the method fix + tests — +call it 1–2 days, and it dissolves issues 1 and 4 together. + +--- + +## 2. Persistence: in-memory context sequence counter — P1 + +### Symptom + +Under Inngest step **replay** (the engine's normal behavior on retry), worker +context events can collide on the unique `(task_attempt_id, sequence)` +constraint or restart their sequence at 0, corrupting the per-execution context +stream. Also a slow memory leak on long-lived API processes. + +### Root cause + +`ContextEventService` assigns `sequence` from a **process-local dict**, not the +DB (`context/service.py:33,39-40,118`). The module docstring states the +load-bearing assumption — "safe because each execution runs in a single Inngest +invocation" (`context/service.py:3-4`) — which Inngest replay does not +guarantee. `_sequence_counters` and `_active_turn_ids` are keyed by +`execution_id` and never popped (unbounded growth). + +It also `session.commit()`s inside the service (`context/service.py:121`), +contradicting the repository-layer "caller owns the transaction" convention +(`test_repository_layer_conventions.py` `test_no_commit_inside_repository`). + +### Scope + +- **method fix** — derive the next sequence from + `SELECT COALESCE(MAX(sequence), -1) + 1 WHERE task_attempt_id = ?` inside the + same session/transaction; drop the in-memory `_sequence_counters`. Resolve + `turn_id` from the persisted tail rather than `_active_turn_ids`, or scope + both dicts to a single `persist`-batch object that is discarded after use. +- **design decision (small)** — confirm commit ownership: either keep the + internal commit and document this service as a transaction owner, or push the + commit to the caller/job to match the convention. +- **tests** — replay simulation: persist N chunks, replay the same chunks, assert + no `IntegrityError` and a contiguous sequence. + +**Effort:** method fix + tests. No schema change. + +--- + +## 3. Sandbox: `BaseSandboxManager.create()` leaks on partial failure — P2 + +### Symptom + +If `_install_dependencies` or `_verify_setup` raises, the E2B sandbox stays +**alive and billing** on the remote and a stale entry remains in the +class-level `_sandboxes` singleton. + +### Root cause + +`create()` registers the sandbox in class-level state and then runs setup steps +with **no cleanup-on-failure** (`manager.py:327-351`). `_create_directory_structure` +*does* wrap-and-terminate on failure earlier in the same method, so the +inconsistency is self-evident in one function. + +### Scope + +This sits on top of a larger structural question, so two paths: + +- **If `BaseSandboxManager` stays** (smoke/materialization path): **method fix** — + wrap the three setup calls in `try/except` that calls `terminate(...)` and + removes the `_sandboxes` entry before re-raising. Mirror the existing + directory-structure handling. + a **test** asserting terminate-on-setup-failure. +- **Bigger question (refactor/deletion):** the live v2 execution path uses + object-bound `Sandbox` + `E2BSandboxRuntime` and never touches this manager, + so we maintain two sandbox stacks. Decide whether to **delete** + `BaseSandboxManager` and migrate smoke/materialization to the v2 runtime. That + is a separate refactor RFC; until then, apply the method fix so the legacy path + doesn't leak. + +**Effort:** method fix is small; the consolidation is a refactor RFC, tracked +separately. + +--- + +## 4. Cross-cutting: the projection↔WAL sync invariant is unenforced — P1 + +### Symptom + +Issue 1 was possible because nothing asserts that projection mutations and WAL +appends stay in lockstep, or that WAL replay reproduces the live projection. + +### Scope — **tests** (+ a small **refactor** to make the invariant hard to break) + +1. **roundtrip test** — materialize a sample, drive a representative sequence of + `RuntimeGraphRepository` mutations (status, field, edges, dynamic spawn), + then assert `reconstruct_sample_runtime_state_at(...)` equals the live + projection (status per node, edge statuses, task snapshots). +2. **mutator-completeness test** — reflect over `RuntimeGraphRepository` public + mutators and assert each appends at least one WAL event (would have failed on + `update_node_field` today). +3. **refactor (optional, recommended)** — route every projection write through a + single private helper that takes the projection change *and* the WAL row + together, so "update projection without WAL" is not expressible. This is the + structural guard that makes issue 1 unrepeatable. + +--- + +## 5. Lower-severity hostility (track, fix opportunistically) + +- **Stringly-typed graph state machine** (`status.py` literals; edges have no FK + to nodes; acyclicity only in `graph_repository._check_no_cycle:544`). Legal + transitions are reconstructed from scattered boolean guards in `lifecycle.py` + / `task_management.py`. **Scope:** design decision (enum + documented + transition table) + refactor; defer unless we touch the scheduler. +- **God modules** — `sample_lifecycle.py` (931, class misnamed `WorkflowService` + mixing orchestration with CLI inspection), `graph_repository.py` (645), + `task_management.py` (634). `lifecycle.py:168` carries a TODO that propagation + "would benefit from being a service or repository method." **Scope:** refactor; + split orchestration from inspection/query helpers. Defer; not a correctness bug. + +--- + +## Summary table + +| # | Issue | Priority | Scope | +|---|-------|----------|-------| +| 1 | `refine_task` edit dropped on rerun | P1 | event-source the fold: design decision + central write helper + method fix + tests (no schema change) | +| 2 | In-memory context sequence counter | P1 | method fix + tests | +| 3 | Sandbox `create()` partial-failure leak | P2 | method fix now; deletion is a separate refactor RFC | +| 4 | Projection↔WAL sync unenforced | P1 | tests + small refactor (single write helper) | +| 5 | Stringly-typed state machine; god modules | P3 | refactor / design decision; defer | + +**Headline answer to "is persistence just a method update?":** No, but it's not +a migration either. The principled fix is to **make current state the fold of +the WAL** (authored base ⊕ typed editable fields ⊕ a namespaced `attributes` +map), with the projection demoted to a derived cache written through a single +helper. That is **one design decision + a small central refactor + a method fix ++ invariant tests** — and it supports mutating arbitrary user attributes without +destroying the timeline. The schema is fine: the dual projection/WAL model is +intentional, event kinds are free-form strings, and the legacy tables are +already gone. diff --git a/docs/superpowers/plans/2026-05-29-real-sandbox-smoke-posture.md b/docs/superpowers/plans/2026-05-29-real-sandbox-smoke-posture.md new file mode 100644 index 000000000..48a9607b3 --- /dev/null +++ b/docs/superpowers/plans/2026-05-29-real-sandbox-smoke-posture.md @@ -0,0 +1,1673 @@ +# Real Sandbox Smoke Posture Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Make `ergon test smoke` exercise the real sandbox class for each benchmark environment instead of the local `SmokePublicSandbox` substitute, while making the canonical smoke scenario realistic and assertion-rich enough to catch runtime, dashboard, RL-view, resource-handoff, messaging, and evaluation-ordering regressions. + +**Architecture:** Keep one top-of-stack PR that changes both smoke posture and smoke depth. Replace the sandbox posture so root and dynamically spawned smoke tasks carry the benchmark environment's production sandbox type (`ResearchE2BSandbox`, `LeanSandbox`, `SWEBenchSandbox`, and `GDPEvalSandbox` where applicable). At the same time, upgrade the deterministic smoke scenario: realistic root/child task prompts, seeded source documents, cross-task resource handoff, message visibility/reply checks, production builtin toolkit calls, synthetic token/logprob telemetry, tighter host-side assertions, and one deterministic dynamic graph/team mutation lane. Most assertions stay static or host-side for debuggability; add exactly one dynamic-subtask evaluator on the handoff consumer to prove intermediary task evaluation works at terminal state without adding evaluators to every child. The smoke workers remain deterministic and do not invoke an LLM, but when they need file/command/report/proof/repo operations they should exercise the real `ergon_builtins` toolkit surfaces. The test harness should fail fast if a smoke task snapshot contains `SmokePublicSandbox` or `TestSandbox`. + +**Tech Stack:** Python 3.13, pytest, FastAPI danger test harness, Ergon public `Sandbox`/`Task` APIs, benchmark sandbox classes, Docker dev stack, Playwright dashboard smoke specs. + +--- + +## File Map + +- Modify `tests/fixtures/smoke_components/benchmarks.py` + - Owns deterministic smoke environments and root task construction. + - Replace `SmokePublicSandbox()` with per-environment real sandbox factories. + - Improve root task names/descriptions/payloads so dashboard/API surfaces resemble realistic benchmark work. + +- Modify `tests/fixtures/smoke_components/smoke_base/dynamic_tasks.py` + - Owns dynamic child task construction for recursive smoke workers. + - Carry the parent's sandbox type/config into spawned child tasks, or accept an explicit sandbox factory/spec. + - Preserve realistic child task descriptions from the smoke graph instead of generic "child task" phrasing. + +- Modify `tests/fixtures/smoke_components/smoke_base/constants.py` + - Owns the recursive child-task graph shape. + - Replace abstract node descriptions with realistic subtasks that require file IO, shell commands, artifact production, resource handoff, messaging, dependency ordering, and nested delegation. + +- Modify `tests/fixtures/smoke_components/smoke_base/worker_base.py` + - Owns parent smoke worker orchestration. + - Seed a realistic source document/message before children run and execute one deterministic graph/team mutation lane. + +- Modify `tests/fixtures/smoke_components/smoke_base/leaf_base.py` + - Owns leaf worker execution. + - Make selected leaves read upstream resources/messages, invoke production builtin toolkits where practical, emit deterministic token/logprob metadata, and produce handoff artifacts. + +- Modify per-environment smoke workers under `tests/fixtures/smoke_components/workers/` + - Bind each environment to the production toolkit it should exercise: + `ResearchRubricsToolkit`, `MiniF2FToolkit`, `SWEBenchToolkit`, and `GDPEvalToolkit` if included. + +- Modify `tests/fixtures/smoke_components/smoke_base/criterion_base.py` + - Owns shared smoke criterion assertions. + - Add exactly one dynamic-subtask evaluator/criterion for the handoff consumer so smoke proves non-root dynamic task evaluation observes terminal state and required upstream resources/messages. + +- Modify `tests/e2e/_asserts.py` + - Owns host-side E2E assertions. + - Add stricter static/host-side checks for resource lineage, messages, RL training records/logprobs, graph/team mutation visibility, and root + dynamic evaluation rows. + +- Modify or retire `tests/fixtures/smoke_components/sandbox.py` + - This file currently implements the fake local smoke sandbox. + - Keep only if still needed for narrow unit tests; canonical E2E smoke must not use it. + +- Modify `tests/e2e/_submit.py` + - If needed, include a real-sandbox posture flag in the danger endpoint payload. + - Default should be real sandbox for `ergon test smoke`. + +- Modify `ergon_core/ergon_core/core/infrastructure/http/routes/test_harness.py` + - Add a smoke posture guard before submission. + - Optionally support the future multi-environment single-experiment submission shape, but keep this PR focused unless the code change is trivial. + +- Add/modify `tests/e2e/test_smoke_real_sandbox_posture.py` + - Fast static/contract test proving smoke fixture task JSON uses real sandbox types and rejects fake test sandboxes. + +- Modify `tests/e2e/test_researchrubrics_smoke.py`, `tests/e2e/test_minif2f_smoke.py`, `tests/e2e/test_swebench_smoke.py` + - Update misleading docstrings/comments that say "against real E2B" only after the posture is actually true. + +--- + +## Task 1: Add A Failing Contract Test For Root Smoke Sandbox Types + +**Files:** +- Create: `tests/e2e/test_smoke_real_sandbox_posture.py` +- Read: `tests/fixtures/smoke_components/benchmarks.py` + +- [ ] **Step 1: Write the failing test** + +Create `tests/e2e/test_smoke_real_sandbox_posture.py`: + +```python +from tests.fixtures.smoke_components.benchmarks import ( + MiniF2FSmokeEnvironment, + ResearchRubricsSmokeEnvironment, + SweBenchSmokeEnvironment, +) + + +def _root_sandbox_type(environment_cls) -> str: + environment = environment_cls() + sample = next(iter(environment.iter_samples())) + [task] = sample.tasks + task_json = task.model_dump(mode="json") + return task_json["sandbox"]["_type"] + + +def test_smoke_root_tasks_use_real_environment_sandboxes() -> None: + assert _root_sandbox_type(ResearchRubricsSmokeEnvironment) == ( + "ergon_builtins.benchmarks.researchrubrics.sandbox:ResearchE2BSandbox" + ) + assert _root_sandbox_type(MiniF2FSmokeEnvironment) == ( + "ergon_builtins.benchmarks.minif2f.sandbox:LeanSandbox" + ) + assert _root_sandbox_type(SweBenchSmokeEnvironment) == ( + "ergon_builtins.benchmarks.swebench_verified.sandbox:SWEBenchSandbox" + ) +``` + +- [ ] **Step 2: Verify RED** + +Run: + +```bash +uv run pytest tests/e2e/test_smoke_real_sandbox_posture.py::test_smoke_root_tasks_use_real_environment_sandboxes -q +``` + +Expected: FAIL because the root smoke environments currently serialize `tests.fixtures.smoke_components.sandbox:SmokePublicSandbox`. + +- [ ] **Step 3: Commit only the failing test if desired** + +For strict TDD checkpoints: + +```bash +git add tests/e2e/test_smoke_real_sandbox_posture.py +git commit -m "test: require real sandbox posture for smoke roots" +``` + +--- + +## Task 2: Switch Root Smoke Environments To Real Sandbox Classes + +**Files:** +- Modify: `tests/fixtures/smoke_components/benchmarks.py` +- Test: `tests/e2e/test_smoke_real_sandbox_posture.py` + +- [ ] **Step 1: Import the real benchmark sandboxes** + +In `tests/fixtures/smoke_components/benchmarks.py`, replace the `SmokePublicSandbox` import with real sandbox imports: + +```python +from ergon_builtins.benchmarks.gdpeval.sandbox import GDPEvalSandbox +from ergon_builtins.benchmarks.minif2f.sandbox import LeanSandbox +from ergon_builtins.benchmarks.researchrubrics.sandbox import ResearchE2BSandbox +from ergon_builtins.benchmarks.swebench_verified.sandbox import SWEBenchSandbox +``` + +- [ ] **Step 2: Replace root sandbox construction** + +In each smoke environment `_tasks()` method, use the real sandbox: + +```python +# ResearchRubricsSmokeEnvironment +sandbox=ResearchE2BSandbox(), + +# MiniF2FSmokeEnvironment +sandbox=LeanSandbox(), + +# SweBenchSmokeEnvironment +sandbox=SWEBenchSandbox(), + +# GDPEvalSmokeEnvironment +sandbox=GDPEvalSandbox(), +``` + +Remove stale comments claiming the smoke fixture uses `SmokeSandboxManager`. + +- [ ] **Step 3: Verify GREEN for root posture** + +Run: + +```bash +uv run pytest tests/e2e/test_smoke_real_sandbox_posture.py::test_smoke_root_tasks_use_real_environment_sandboxes -q +``` + +Expected: PASS. + +- [ ] **Step 4: Commit** + +```bash +git add tests/fixtures/smoke_components/benchmarks.py tests/e2e/test_smoke_real_sandbox_posture.py +git commit -m "test: run smoke roots with real benchmark sandboxes" +``` + +--- + +## Task 3: Ensure Dynamically Spawned Smoke Tasks Keep Real Sandbox Posture + +**Files:** +- Modify: `tests/fixtures/smoke_components/smoke_base/dynamic_tasks.py` +- Modify if needed: `tests/fixtures/smoke_components/smoke_base/worker_base.py` +- Modify if needed: `tests/fixtures/smoke_components/smoke_base/recursive.py` +- Test: `tests/e2e/test_smoke_real_sandbox_posture.py` + +- [ ] **Step 1: Add failing dynamic-task posture test** + +Append to `tests/e2e/test_smoke_real_sandbox_posture.py`: + +```python +from ergon_core.core.persistence.shared.types import AssignedWorkerSlug, TaskSlug +from tests.fixtures.smoke_components.smoke_base.dynamic_tasks import ( + SmokeChildTaskSpec, + smoke_task_from_spec, +) + + +def _child_sandbox_type(environment_cls) -> str: + environment = environment_cls() + sample = next(iter(environment.iter_samples())) + [parent_task] = sample.tasks + child = smoke_task_from_spec( + parent_task=parent_task, + spec=SmokeChildTaskSpec( + task_slug=TaskSlug("child"), + description="child task", + assigned_worker_slug=AssignedWorkerSlug("child-worker"), + depends_on=[], + ), + model="openai:gpt-4o", + ) + return child.model_dump(mode="json")["sandbox"]["_type"] + + +def test_smoke_dynamic_tasks_inherit_real_parent_sandbox_type() -> None: + assert _child_sandbox_type(ResearchRubricsSmokeEnvironment) == ( + "ergon_builtins.benchmarks.researchrubrics.sandbox:ResearchE2BSandbox" + ) + assert _child_sandbox_type(MiniF2FSmokeEnvironment) == ( + "ergon_builtins.benchmarks.minif2f.sandbox:LeanSandbox" + ) + assert _child_sandbox_type(SweBenchSmokeEnvironment) == ( + "ergon_builtins.benchmarks.swebench_verified.sandbox:SWEBenchSandbox" + ) +``` + +- [ ] **Step 2: Verify RED** + +Run: + +```bash +uv run pytest tests/e2e/test_smoke_real_sandbox_posture.py::test_smoke_dynamic_tasks_inherit_real_parent_sandbox_type -q +``` + +Expected: FAIL if `smoke_task_from_spec` still constructs `SmokePublicSandbox()`. + +- [ ] **Step 3: Implement inheritance from parent task sandbox** + +In `tests/fixtures/smoke_components/smoke_base/dynamic_tasks.py`, replace direct fake sandbox construction with a copy of the parent task sandbox config. + +Preferred implementation: + +```python +from copy import deepcopy + + +def _clone_parent_sandbox(parent_task: Task) -> Sandbox: + return deepcopy(parent_task.sandbox) +``` + +Then in `smoke_task_from_spec(...)`, set: + +```python +sandbox=_clone_parent_sandbox(parent_task), +``` + +If `Task.sandbox` may be absent, fail loudly: + +```python +if parent_task.sandbox is None: + raise ValueError("smoke dynamic tasks require parent_task.sandbox") +``` + +- [ ] **Step 4: Verify dynamic posture** + +Run: + +```bash +uv run pytest tests/e2e/test_smoke_real_sandbox_posture.py -q +``` + +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add tests/fixtures/smoke_components/smoke_base/dynamic_tasks.py tests/e2e/test_smoke_real_sandbox_posture.py +git commit -m "test: inherit real sandbox posture for smoke subtasks" +``` + +--- + +## Task 4: Make Smoke Task Names And Descriptions Realistic + +**Files:** +- Modify: `tests/fixtures/smoke_components/benchmarks.py` +- Modify: `tests/fixtures/smoke_components/smoke_base/constants.py` +- Modify if needed: `tests/fixtures/smoke_components/smoke_base/dynamic_tasks.py` +- Test: `tests/e2e/test_smoke_real_sandbox_posture.py` + +- [ ] **Step 1: Add failing realism contract tests** + +Append to `tests/e2e/test_smoke_real_sandbox_posture.py`: + +```python +from tests.fixtures.smoke_components.smoke_base.constants import SUBTASK_GRAPH + + +def _root_task_snapshot(environment_cls) -> dict: + environment = environment_cls() + sample = next(iter(environment.iter_samples())) + [task] = sample.tasks + return task.model_dump(mode="json") + + +def test_smoke_root_tasks_have_realistic_environment_descriptions() -> None: + research = _root_task_snapshot(ResearchRubricsSmokeEnvironment) + assert research["task_slug"] == "research-report" + assert "research brief" in research["description"].lower() + assert "evidence" in research["description"].lower() + + mini = _root_task_snapshot(MiniF2FSmokeEnvironment) + assert mini["task_slug"] == "lean-proof" + assert "lean" in mini["description"].lower() + assert "proof" in mini["description"].lower() + + swe = _root_task_snapshot(SweBenchSmokeEnvironment) + assert swe["task_slug"] == "repo-fix" + assert "repository" in swe["description"].lower() + assert "regression" in swe["description"].lower() + + +def test_smoke_child_task_descriptions_cover_real_runtime_behaviors() -> None: + descriptions = " ".join(description.lower() for _, _, description in SUBTASK_GRAPH) + for expected in ( + "inspect", + "write", + "command", + "artifact", + "verify", + "summarize", + ): + assert expected in descriptions +``` + +- [ ] **Step 2: Verify RED** + +Run: + +```bash +uv run pytest tests/e2e/test_smoke_real_sandbox_posture.py::test_smoke_root_tasks_have_realistic_environment_descriptions tests/e2e/test_smoke_real_sandbox_posture.py::test_smoke_child_task_descriptions_cover_real_runtime_behaviors -q +``` + +Expected: FAIL because current smoke roots use generic slugs/descriptions like `smoke-001`, `mathd_algebra_478`, and child graph descriptions like `Nested line node 2a`. + +- [ ] **Step 3: Update root task slugs/descriptions** + +In `tests/fixtures/smoke_components/benchmarks.py`, update the smoke environment constants: + +```python +class ResearchRubricsSmokeEnvironment(_SingleTaskSmokeEnvironment): + task_slug: ClassVar[str] = "research-report" + task_description: ClassVar[str] = ( + "Draft a concise research brief from the supplied prompt, preserve the " + "expected smoke-test evidence marker, and write the final report artifact." + ) + + +class MiniF2FSmokeEnvironment(_SingleTaskSmokeEnvironment): + task_slug: ClassVar[str] = "lean-proof" + task_description: ClassVar[str] = ( + "Inspect the Lean theorem statement, run the proof-check command, and " + "produce a final proof artifact for the smoke theorem." + ) + + +class SweBenchSmokeEnvironment(_SingleTaskSmokeEnvironment): + task_slug: ClassVar[str] = "repo-fix" + task_description: ClassVar[str] = ( + "Inspect the repository regression, apply a minimal code fix, run the " + "verification command, and summarize the patch artifact." + ) +``` + +For `GDPEvalSmokeEnvironment`, use: + +```python +task_slug: ClassVar[str] = "workflow-debug" +task_description: ClassVar[str] = ( + "Inspect the workflow inputs, run the smoke validation command, and produce " + "a final workflow-debug artifact." +) +``` + +- [ ] **Step 4: Update child graph slugs and descriptions** + +In `tests/fixtures/smoke_components/smoke_base/constants.py`, rename the dynamic topology slugs to semantic IDs and replace descriptions with realistic work. The slugs are part of the smoke contract and should be referenced through named constants, not repeated as raw strings elsewhere: + +```python +SOURCE_REVIEW_SLUG = "source-review" +ENV_PROBE_SLUG = "environment-probe" +HANDOFF_VERIFY_SLUG = "handoff-verify" +PRIMARY_ARTIFACT_SLUG = "primary-artifact" +EVIDENCE_ARTIFACT_SLUG = "evidence-artifact" +ARTIFACT_SUMMARY_SLUG = "artifact-summary" +METADATA_REVIEW_SLUG = "metadata-review" +METADATA_VALIDATE_SLUG = "metadata-validate" +COMPLETION_MARKER_SLUG = "completion-marker" + +EXPECTED_SUBTASK_SLUGS: tuple[str, ...] = ( + SOURCE_REVIEW_SLUG, + ENV_PROBE_SLUG, + HANDOFF_VERIFY_SLUG, + PRIMARY_ARTIFACT_SLUG, + EVIDENCE_ARTIFACT_SLUG, + ARTIFACT_SUMMARY_SLUG, + METADATA_REVIEW_SLUG, + METADATA_VALIDATE_SLUG, + COMPLETION_MARKER_SLUG, +) + +SUBTASK_GRAPH: tuple[tuple[str, tuple[str, ...], str], ...] = ( + (SOURCE_REVIEW_SLUG, (), "Inspect the task payload and write an initial workspace note."), + (ENV_PROBE_SLUG, (), "Run a sandbox command to probe the environment and record the result."), + ( + HANDOFF_VERIFY_SLUG, + (SOURCE_REVIEW_SLUG, ENV_PROBE_SLUG), + "Verify the inspected inputs against the command probe output.", + ), + ( + PRIMARY_ARTIFACT_SLUG, + (HANDOFF_VERIFY_SLUG,), + "Write the primary artifact into the final output directory.", + ), + ( + EVIDENCE_ARTIFACT_SLUG, + (HANDOFF_VERIFY_SLUG,), + "Create a secondary evidence artifact for evaluation.", + ), + ( + ARTIFACT_SUMMARY_SLUG, + (PRIMARY_ARTIFACT_SLUG, EVIDENCE_ARTIFACT_SLUG), + "Summarize the artifact set and dependency decisions.", + ), + (METADATA_REVIEW_SLUG, (), "Inspect optional metadata and produce an auxiliary note."), + ( + METADATA_VALIDATE_SLUG, + (METADATA_REVIEW_SLUG,), + "Validate the auxiliary note with a sandbox command.", + ), + (COMPLETION_MARKER_SLUG, (), "Write a standalone completion marker artifact."), +) +``` + +If `NESTED_SUBTASK_GRAPH` in `tests/fixtures/smoke_components/smoke_base/recursive.py` still uses generic descriptions, update it to: + +```python +NESTED_INSPECT_SLUG = "nested-input-review" +NESTED_VERIFY_SLUG = "nested-verification" +NESTED_LINE_SLUGS: tuple[str, ...] = (NESTED_INSPECT_SLUG, NESTED_VERIFY_SLUG) + +NESTED_SUBTASK_GRAPH: tuple[tuple[str, tuple[str, ...], str], ...] = ( + (NESTED_INSPECT_SLUG, (), "Inspect nested worker inputs and write a nested scratch note."), + ( + NESTED_VERIFY_SLUG, + (NESTED_INSPECT_SLUG,), + "Run a nested verification command and summarize the result.", + ), +) +``` + +Update any tests/assertions that import `NESTED_LINE_SLUGS` to import it from the canonical constants module if the ownership moves. The old internal slugs (`d_root`, `d_left`, `l_2`, etc.) should not remain in canonical smoke assertions. + +- [ ] **Step 5: Verify realism tests** + +Run: + +```bash +uv run pytest tests/e2e/test_smoke_real_sandbox_posture.py -q +``` + +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add tests/fixtures/smoke_components/benchmarks.py tests/fixtures/smoke_components/smoke_base/constants.py tests/fixtures/smoke_components/smoke_base/recursive.py tests/e2e/test_smoke_real_sandbox_posture.py +git commit -m "test: make smoke task prompts more realistic" +``` + +--- + +## Task 5: Add A Harness Guard Against Fake Sandboxes In Canonical Smoke + +**Files:** +- Modify: `ergon_core/ergon_core/core/infrastructure/http/routes/test_harness.py` +- Test: `tests/e2e/test_smoke_real_sandbox_posture.py` + +- [ ] **Step 1: Add failing guard test** + +Append to `tests/e2e/test_smoke_real_sandbox_posture.py`: + +```python +import pytest + +from ergon_core.core.infrastructure.http.routes.test_harness import ( + _assert_real_smoke_sandbox_posture, +) + + +def test_test_harness_rejects_smoke_public_sandbox() -> None: + bad_task_json = { + "sandbox": { + "_type": "tests.fixtures.smoke_components.sandbox:SmokePublicSandbox", + } + } + with pytest.raises(ValueError, match="canonical smoke requires real benchmark sandboxes"): + _assert_real_smoke_sandbox_posture([bad_task_json]) + + +def test_test_harness_rejects_test_support_sandbox() -> None: + bad_task_json = { + "sandbox": { + "_type": "ergon_core.test_support.task_factory:TestSandbox", + } + } + with pytest.raises(ValueError, match="canonical smoke requires real benchmark sandboxes"): + _assert_real_smoke_sandbox_posture([bad_task_json]) +``` + +- [ ] **Step 2: Verify RED** + +Run: + +```bash +uv run pytest tests/e2e/test_smoke_real_sandbox_posture.py::test_test_harness_rejects_smoke_public_sandbox tests/e2e/test_smoke_real_sandbox_posture.py::test_test_harness_rejects_test_support_sandbox -q +``` + +Expected: FAIL because `_assert_real_smoke_sandbox_posture` does not exist. + +- [ ] **Step 3: Implement the guard helper** + +In `ergon_core/ergon_core/core/infrastructure/http/routes/test_harness.py`, add: + +```python +_DISALLOWED_SMOKE_SANDBOX_TYPES = frozenset( + { + "tests.fixtures.smoke_components.sandbox:SmokePublicSandbox", + "ergon_core.test_support.task_factory:TestSandbox", + } +) + + +def _assert_real_smoke_sandbox_posture(task_jsons: list[dict]) -> None: + for task_json in task_jsons: + sandbox = task_json.get("sandbox") + sandbox_type = sandbox.get("_type") if isinstance(sandbox, dict) else None + if sandbox_type in _DISALLOWED_SMOKE_SANDBOX_TYPES: + raise ValueError( + "canonical smoke requires real benchmark sandboxes; " + f"found disallowed sandbox type {sandbox_type!r}" + ) +``` + +- [ ] **Step 4: Call the guard before submit** + +In `submit_experiment_samples`, after materializing `samples` and before `experiment.submit(...)`, call: + +```python +samples = list(environment.iter_samples())[: body.limit] +_assert_real_smoke_sandbox_posture( + [task.model_dump(mode="json") for sample in samples for task in sample.tasks] +) +experiment = Experiment( + name=body.experiment, + environments=[ + _HarnessEnvironment( + name=body.environment_slug, + samples=samples, + metadata={"source": "test-harness"}, + ) + ], + metadata={"source": "test-harness"}, +) +``` + +Translate `ValueError` to `HTTPException(500, ...)` or let it raise as a test-harness bug. Prefer a 500 because this is a broken harness invariant, not user input. + +- [ ] **Step 5: Verify guard tests** + +Run: + +```bash +uv run pytest tests/e2e/test_smoke_real_sandbox_posture.py -q +``` + +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add ergon_core/ergon_core/core/infrastructure/http/routes/test_harness.py tests/e2e/test_smoke_real_sandbox_posture.py +git commit -m "test: reject fake sandboxes in smoke harness" +``` + +--- + +## Task 6: Seed Realistic Documents And Cross-Task Handoffs + +**Files:** +- Modify: `tests/fixtures/smoke_components/smoke_base/worker_base.py` +- Modify: `tests/fixtures/smoke_components/smoke_base/leaf_base.py` +- Modify: `tests/fixtures/smoke_components/smoke_base/criterion_base.py` +- Modify: `tests/e2e/_asserts.py` +- Test: `tests/e2e/test_smoke_real_sandbox_posture.py` +- Test: `tests/e2e/test_researchrubrics_smoke.py`, `tests/e2e/test_minif2f_smoke.py`, `tests/e2e/test_swebench_smoke.py` + +- [ ] **Step 1: Add failing semantic coverage tests** + +Append to `tests/e2e/test_smoke_real_sandbox_posture.py`: + +```python +from tests.fixtures.smoke_components.smoke_base.constants import ( + EXPECTED_RESOURCE_HANDOFF, + HANDOFF_RESOURCE_NAME, + HANDOFF_VERIFY_SLUG, + SEEDED_SOURCE_DOC_NAME, + SMOKE_THREAD_TOPIC, + SOURCE_REVIEW_SLUG, +) + + +def test_smoke_semantic_contract_names_seeded_docs_handoffs_and_threads() -> None: + assert SEEDED_SOURCE_DOC_NAME == "smoke_source_brief.md" + assert SMOKE_THREAD_TOPIC == "smoke-coordination" + assert SOURCE_REVIEW_SLUG == "source-review" + assert HANDOFF_VERIFY_SLUG == "handoff-verify" + assert HANDOFF_RESOURCE_NAME == "handoff_source_review.json" + assert EXPECTED_RESOURCE_HANDOFF.producer_slug == SOURCE_REVIEW_SLUG + assert EXPECTED_RESOURCE_HANDOFF.consumer_slug == HANDOFF_VERIFY_SLUG + assert EXPECTED_RESOURCE_HANDOFF.resource_name == HANDOFF_RESOURCE_NAME +``` + +Expected new constants: + +```python +from dataclasses import dataclass + + +@dataclass(frozen=True) +class ExpectedResourceHandoff: + producer_slug: str + consumer_slug: str + resource_name: str + + +SEEDED_SOURCE_DOC_NAME = "smoke_source_brief.md" +SMOKE_THREAD_TOPIC = "smoke-coordination" +HANDOFF_RESOURCE_NAME = "handoff_source_review.json" +EXPECTED_RESOURCE_HANDOFF = ExpectedResourceHandoff( + producer_slug=SOURCE_REVIEW_SLUG, + consumer_slug=HANDOFF_VERIFY_SLUG, + resource_name=HANDOFF_RESOURCE_NAME, +) +``` + +- [ ] **Step 2: Verify RED** + +Run: + +```bash +uv run pytest tests/e2e/test_smoke_real_sandbox_posture.py::test_smoke_semantic_contract_names_seeded_docs_handoffs_and_threads -q +``` + +Expected: FAIL because those constants do not exist. + +- [ ] **Step 3: Add smoke semantic constants** + +In `tests/fixtures/smoke_components/smoke_base/constants.py`, add the dataclass and constants from Step 1. Keep names stable because both fixture code and host-side assertions will consume them. + +- [ ] **Step 4: Seed a realistic source document and coordination message** + +In `tests/fixtures/smoke_components/smoke_base/worker_base.py`, before spawning children, write a source document and coordination message. Use existing runtime/context APIs already used in smoke workers; the implementation must be deterministic. + +Target behavior: + +```python +seeded_doc = "\n".join( + [ + "# Smoke Source Brief", + "", + "The child tasks must preserve marker SMOKE_RESOURCE_HANDOFF_OK.", + f"The consumer task must read {HANDOFF_RESOURCE_NAME} before writing its artifact.", + ] +) +await task.sandbox.write_file(f"/workspace/{SEEDED_SOURCE_DOC_NAME}", seeded_doc) +await communication_service.save_message( + CreateMessageRequest( + sample_id=context.sample_id, + task_attempt_id=context.execution_id, + from_agent_id="parent", + to_agent_id=EXPECTED_RESOURCE_HANDOFF.consumer_slug, + thread_topic=SMOKE_THREAD_TOPIC, + content=f"Read {HANDOFF_RESOURCE_NAME} before producing your artifact.", + ) +) +``` + +If `WorkerContext` already provides a higher-level message/resource API, use that instead of directly importing `communication_service`. The assertion contract is the important bit: the message must exist before `EXPECTED_RESOURCE_HANDOFF.consumer_slug` starts. + +- [ ] **Step 5: Make the source-review task produce a named handoff resource** + +In `tests/fixtures/smoke_components/smoke_base/leaf_base.py`, when executing `EXPECTED_RESOURCE_HANDOFF.producer_slug`, write and persist `HANDOFF_RESOURCE_NAME` with deterministic content: + +```json +{ + "marker": "SMOKE_RESOURCE_HANDOFF_OK", + "producer": "source-review", + "consumer": "handoff-verify", + "source_doc": "smoke_source_brief.md" +} +``` + +The resource must be persisted as a normal task resource, not only left in the sandbox filesystem. + +- [ ] **Step 6: Make the handoff-verify task read the handoff and reply on the thread** + +In `leaf_base.py`, when executing `EXPECTED_RESOURCE_HANDOFF.consumer_slug`: + +```python +handoff = await context.read_resource_by_name(HANDOFF_RESOURCE_NAME) +assert "SMOKE_RESOURCE_HANDOFF_OK" in handoff.decode("utf-8") +await communication_service.save_message( + CreateMessageRequest( + sample_id=context.sample_id, + task_attempt_id=context.execution_id, + from_agent_id=EXPECTED_RESOURCE_HANDOFF.consumer_slug, + to_agent_id="parent", + thread_topic=SMOKE_THREAD_TOPIC, + content=f"Confirmed {HANDOFF_RESOURCE_NAME} marker SMOKE_RESOURCE_HANDOFF_OK.", + ) +) +``` + +If no `read_resource_by_name` helper exists, use the repository/read-service path that production criterion code uses to read task resources. Do not fake success by sharing an in-memory variable. + +- [ ] **Step 7: Add one dynamic-subtask evaluator for intermediary terminal-state coverage** + +Add one evaluator/criterion attached to `EXPECTED_RESOURCE_HANDOFF.consumer_slug` only. Do not add evaluators to every dynamic child. The purpose is narrow: prove evaluation works for a non-root dynamic task after that task reaches terminal state, and prove the evaluator context can see the upstream handoff resource and coordination messages. + +Target criterion behavior: + +```python +class SmokeIntermediateHandoffCriterion(Criterion): + slug = "smoke-intermediate-handoff" + + async def evaluate(self, context: CriterionContext) -> CriterionResult: + task_slug = context.task.task_slug + assert task_slug == EXPECTED_RESOURCE_HANDOFF.consumer_slug + handoff_body = read_handoff_resource(context, HANDOFF_RESOURCE_NAME) + assert handoff_body["marker"] == "SMOKE_RESOURCE_HANDOFF_OK" + assert handoff_body["producer"] == EXPECTED_RESOURCE_HANDOFF.producer_slug + assert handoff_body["consumer"] == EXPECTED_RESOURCE_HANDOFF.consumer_slug + assert thread_contains( + context, + topic=SMOKE_THREAD_TOPIC, + from_agent_id="parent", + to_agent_id=EXPECTED_RESOURCE_HANDOFF.consumer_slug, + ) + assert thread_contains( + context, + topic=SMOKE_THREAD_TOPIC, + from_agent_id=EXPECTED_RESOURCE_HANDOFF.consumer_slug, + content_fragment=f"Confirmed {HANDOFF_RESOURCE_NAME}", + ) + return CriterionResult(score=1.0, reason="intermediate handoff visible") +``` + +Attach this criterion to the `handoff-verify` dynamic task when it is created. If the current dynamic-task API does not support attaching an evaluator at spawn time, extend the smoke task construction path just enough to pass `evaluators=[SmokeIntermediateHandoffCriterion(...)]` for `EXPECTED_RESOURCE_HANDOFF.consumer_slug`. + +This makes exactly one check happen through the runtime evaluation path, not just host-side pytest. It covers the ordering gap we care about: dynamic task evaluation should happen after that dynamic task is terminal and after the required upstream resource/message exists. + +- [ ] **Step 8: Add host-side E2E assertions** + +In `tests/e2e/_asserts.py`, add: + +```python +def _assert_resource_handoff_and_coordination_thread(sample_id: UUID) -> None: + handoff_prefix = HANDOFF_RESOURCE_NAME.removesuffix(".json") + handoffs = list_named_resources(sample_id, prefix=handoff_prefix, suffix=".json") + assert len(handoffs) == 1, ( + f"expected one {EXPECTED_RESOURCE_HANDOFF.producer_slug} handoff resource, " + f"got {len(handoffs)}" + ) + body = json.loads(read_resource_bytes(handoffs[0]).decode("utf-8")) + assert body["marker"] == "SMOKE_RESOURCE_HANDOFF_OK" + assert body["producer"] == EXPECTED_RESOURCE_HANDOFF.producer_slug + assert body["consumer"] == EXPECTED_RESOURCE_HANDOFF.consumer_slug + + snapshot = require_run_snapshot(sample_id) + thread = next((t for t in snapshot.threads if t.topic == SMOKE_THREAD_TOPIC), None) + assert thread is not None, f"missing {SMOKE_THREAD_TOPIC} thread" + messages = sorted(thread.messages, key=lambda message: message.sequence_num) + assert messages[0].from_agent_id == "parent" + assert messages[0].to_agent_id == EXPECTED_RESOURCE_HANDOFF.consumer_slug + assert any( + message.from_agent_id == EXPECTED_RESOURCE_HANDOFF.consumer_slug + and f"Confirmed {HANDOFF_RESOURCE_NAME}" in message.content + for message in messages + ) +``` + +Add a host-side evaluation assertion alongside this: + +```python +def _assert_intermediate_handoff_evaluation(sample_id: UUID) -> None: + snapshot = require_run_snapshot(sample_id) + consumer = next( + task + for task in snapshot.tasks.values() + if task.name == EXPECTED_RESOURCE_HANDOFF.consumer_slug + ) + evaluation = snapshot.evaluations_by_task.get(consumer.id) + assert evaluation is not None, ( + f"expected one intermediate evaluation for {EXPECTED_RESOURCE_HANDOFF.consumer_slug}" + ) + assert evaluation.normalized_score == 1.0 + assert evaluation.evaluator_name + assert any( + criterion.criterion_slug == "smoke-intermediate-handoff" + and criterion.status == "passed" + for criterion in evaluation.criterion_results + ) +``` + +Call `_assert_resource_handoff_and_coordination_thread(rid)` and `_assert_intermediate_handoff_evaluation(rid)` from every happy-path smoke assertion. For the sad path, assert that handoff exists only if `EXPECTED_RESOURCE_HANDOFF.producer_slug` completed and that no blocked task fabricated a reply or successful intermediate evaluation. + +- [ ] **Step 9: Verify focused and E2E tests** + +Run: + +```bash +uv run pytest tests/e2e/test_smoke_real_sandbox_posture.py -q +ergon test smoke +``` + +Expected: PASS. The smoke run should now fail if resource visibility, message persistence, or evaluation access to resources regresses. + +- [ ] **Step 10: Commit** + +```bash +git add tests/fixtures/smoke_components/smoke_base/constants.py tests/fixtures/smoke_components/smoke_base/worker_base.py tests/fixtures/smoke_components/smoke_base/leaf_base.py tests/fixtures/smoke_components/smoke_base/criterion_base.py tests/e2e/_asserts.py tests/e2e/test_smoke_real_sandbox_posture.py +git commit -m "test: add smoke resource handoff and coordination assertions" +``` + +--- + +## Task 7: Exercise Production Builtin Toolkits In Smoke Workers + +**Files:** +- Modify: `tests/fixtures/smoke_components/smoke_base/leaf_base.py` +- Modify: `tests/fixtures/smoke_components/workers/researchrubrics_smoke.py` +- Modify: `tests/fixtures/smoke_components/workers/minif2f_smoke.py` +- Modify: `tests/fixtures/smoke_components/workers/swebench_smoke.py` +- Modify if included: `tests/fixtures/smoke_components/benchmarks.py` +- Modify: `tests/e2e/_asserts.py` +- Test: `tests/e2e/test_smoke_real_sandbox_posture.py` + +- [ ] **Step 1: Add failing toolkit posture contract** + +Append to `tests/e2e/test_smoke_real_sandbox_posture.py`: + +```python +from tests.fixtures.smoke_components.workers.minif2f_smoke import MiniF2FSmokeLeafWorker +from tests.fixtures.smoke_components.workers.researchrubrics_smoke import ( + ResearchRubricsSmokeLeafWorker, +) +from tests.fixtures.smoke_components.workers.swebench_smoke import SweBenchSmokeLeafWorker + + +def test_smoke_leaf_workers_bind_production_builtin_toolkits() -> None: + assert ResearchRubricsSmokeLeafWorker.toolkit_type == ( + "ergon_builtins.benchmarks.researchrubrics.toolkit:ResearchRubricsToolkit" + ) + assert MiniF2FSmokeLeafWorker.toolkit_type == ( + "ergon_builtins.benchmarks.minif2f.toolkit:MiniF2FToolkit" + ) + assert SweBenchSmokeLeafWorker.toolkit_type == ( + "ergon_builtins.benchmarks.swebench_verified.toolkit:SWEBenchToolkit" + ) +``` + +- [ ] **Step 2: Verify RED** + +Run: + +```bash +uv run pytest tests/e2e/test_smoke_real_sandbox_posture.py::test_smoke_leaf_workers_bind_production_builtin_toolkits -q +``` + +Expected: FAIL because smoke leaf workers do not expose production toolkit bindings yet. + +- [ ] **Step 3: Add toolkit bindings to smoke leaf workers** + +In each env worker module, add class attributes on the leaf worker: + +```python +# tests/fixtures/smoke_components/workers/researchrubrics_smoke.py +from ergon_builtins.benchmarks.researchrubrics.toolkit import ResearchRubricsToolkit + + +class ResearchRubricsSmokeLeafWorker(BaseSmokeLeafWorker): + toolkit_cls: ClassVar[type] = ResearchRubricsToolkit + toolkit_type: ClassVar[str] = ( + "ergon_builtins.benchmarks.researchrubrics.toolkit:ResearchRubricsToolkit" + ) +``` + +```python +# tests/fixtures/smoke_components/workers/minif2f_smoke.py +from ergon_builtins.benchmarks.minif2f.toolkit import MiniF2FToolkit + + +class MiniF2FSmokeLeafWorker(BaseSmokeLeafWorker): + toolkit_cls: ClassVar[type] = MiniF2FToolkit + toolkit_type: ClassVar[str] = ( + "ergon_builtins.benchmarks.minif2f.toolkit:MiniF2FToolkit" + ) +``` + +```python +# tests/fixtures/smoke_components/workers/swebench_smoke.py +from ergon_builtins.benchmarks.swebench_verified.toolkit import SWEBenchToolkit + + +class SweBenchSmokeLeafWorker(BaseSmokeLeafWorker): + toolkit_cls: ClassVar[type] = SWEBenchToolkit + toolkit_type: ClassVar[str] = ( + "ergon_builtins.benchmarks.swebench_verified.toolkit:SWEBenchToolkit" + ) +``` + +If GDPEval smoke is wired into `ergon test smoke`, mirror this with `GDPEvalToolkit`. + +- [ ] **Step 4: Invoke toolkit tools deterministically** + +In `tests/fixtures/smoke_components/smoke_base/leaf_base.py`, add a helper that builds tools from `self.toolkit_cls()` and dispatches by toolkit class name. Keep the sequence deterministic and do not invoke a model/ReAct loop. + +Use toolkit name/class as the primary discriminator, not "does this toolkit happen to expose tool names X and Y". Tool-name probing is too brittle: a harmless rename or an added compatibility alias could silently change which branch the smoke takes. The smoke should fail clearly if a known environment is bound to the wrong toolkit, and only then use the small set of tool names needed for that toolkit's intended probe. + +Target helper shape: + +```python +async def _run_builtin_toolkit_probe(self, task: Task, context: WorkerContext) -> dict: + toolkit_cls = getattr(type(self), "toolkit_cls", None) + if toolkit_cls is None: + return {"toolkit": None, "tools": []} + + toolkit = toolkit_cls() + toolkit_name = type(toolkit).__name__ + tools = {tool.name: tool for tool in toolkit.tools(task.sandbox, task)} + + if toolkit_name == "ResearchRubricsToolkit": + write = await tools["write_report"].function( + "final_output/toolkit_probe.md", + "SMOKE_TOOLKIT_PROBE_OK research report", + ) + read = await tools["read_report"].function("final_output/toolkit_probe.md") + return {"toolkit": toolkit_name, "write": write.model_dump(), "read": read.model_dump()} + + if toolkit_name == "MiniF2FToolkit": + path = "/workspace/final_output/toolkit_probe.lean" + content = "theorem smoke_toolkit_probe : True := by trivial\n" + write = await tools["write_lean_file"].function(path, content) + verify = await tools["verify_lean_proof"].function(path) + return {"toolkit": toolkit_name, "write": write.model_dump(), "verify": verify.model_dump()} + + if toolkit_name == "SWEBenchToolkit": + create = await tools["str_replace_editor"].function( + command="create", + path="toolkit_probe.py", + file_text="def add(a, b):\n return a + b\n", + ) + bash = await tools["bash"].function("python -m py_compile toolkit_probe.py", timeout_sec=60) + return {"toolkit": toolkit_name, "create": create.model_dump(), "bash": bash.model_dump()} + + raise AssertionError(f"Unexpected smoke toolkit {toolkit_name!r}") +``` + +Adjust invocation syntax to match `pydantic_ai.tools.Tool`'s callable API if `.function` is not public in the installed version. The important contract is: verify the smoke leaf is bound to the expected production toolkit, use the production toolkit's `tools(...)` method, and call the same live tool function that a worker would expose to an agent. + +- [ ] **Step 5: Persist toolkit probe output as a normal smoke artifact** + +When the helper runs, persist the result as a task resource named: + +```text +toolkit_probe_.json +``` + +The JSON body must include: + +```json +{ + "toolkit": "ResearchRubricsToolkit | MiniF2FToolkit | SWEBenchToolkit", + "marker": "SMOKE_TOOLKIT_PROBE_OK", + "tool_names": ["..."] +} +``` + +- [ ] **Step 6: Add host-side toolkit assertions** + +In `tests/e2e/_asserts.py`, add: + +```python +def _assert_builtin_toolkit_probe(sample_id: UUID, *, expected_toolkit: str) -> None: + resources = list_named_resources(sample_id, prefix="toolkit_probe_", suffix=".json") + assert resources, "expected at least one toolkit probe resource" + bodies = [json.loads(read_resource_bytes(resource).decode("utf-8")) for resource in resources] + matching = [body for body in bodies if body.get("toolkit") == expected_toolkit] + assert matching, f"missing toolkit probe for {expected_toolkit}; saw {bodies}" + assert any(body.get("marker") == "SMOKE_TOOLKIT_PROBE_OK" for body in matching) +``` + +Wire expected toolkit by environment: + +```python +_assert_builtin_toolkit_probe(rid, expected_toolkit="ResearchRubricsToolkit") +_assert_builtin_toolkit_probe(rid, expected_toolkit="MiniF2FToolkit") +_assert_builtin_toolkit_probe(rid, expected_toolkit="SWEBenchToolkit") +``` + +- [ ] **Step 7: Verify** + +Run: + +```bash +uv run pytest tests/e2e/test_smoke_real_sandbox_posture.py -q +ergon test smoke +``` + +Expected: PASS. Failures should point to real builtin toolkit integration issues: serialization, live sandbox attachment, command wrapping, path assumptions, or response-model shape drift. + +- [ ] **Step 8: Commit** + +```bash +git add tests/fixtures/smoke_components/smoke_base/leaf_base.py tests/fixtures/smoke_components/workers/researchrubrics_smoke.py tests/fixtures/smoke_components/workers/minif2f_smoke.py tests/fixtures/smoke_components/workers/swebench_smoke.py tests/e2e/_asserts.py tests/e2e/test_smoke_real_sandbox_posture.py +git commit -m "test: exercise builtin toolkits in smoke workers" +``` + +--- + +## Task 8: Tighten RL View Shape, Evaluation Ordering, And Dynamic Mutation Assertions + +**Files:** +- Modify: `tests/fixtures/smoke_components/smoke_base/leaf_base.py` +- Modify: `tests/fixtures/smoke_components/smoke_base/worker_base.py` +- Modify: `tests/e2e/_asserts.py` +- Modify if needed: `ergon_core/ergon_core/test_support/e2e_read_helpers.py` +- Test: `tests/e2e/test_smoke_real_sandbox_posture.py` +- Test: `ergon_core/tests/unit/views/rl/` + +- [ ] **Step 1: Add failing static contract for synthetic RL metadata** + +Append to `tests/e2e/test_smoke_real_sandbox_posture.py`: + +```python +from tests.fixtures.smoke_components.smoke_base.constants import ( + EXPECTED_PARENT_VISIBLE_CHILD_RESULT, + EXPECTED_SMOKE_LOGPROBS, + EXPECTED_SMOKE_TOKEN_IDS, + EXPECTED_SUBAGENT_INTERNAL_MARKER, +) + + +def test_smoke_contract_declares_synthetic_rl_tokens_and_logprobs() -> None: + assert EXPECTED_SMOKE_TOKEN_IDS == [101, 202, 303] + assert EXPECTED_SMOKE_LOGPROBS == [-0.10, -0.20, -0.30] + assert EXPECTED_SUBAGENT_INTERNAL_MARKER == "SMOKE_CHILD_INTERNAL_THINKING" + assert EXPECTED_PARENT_VISIBLE_CHILD_RESULT == "SMOKE_CHILD_RESULT_PARENT_VISIBLE" +``` + +- [ ] **Step 2: Verify RED** + +Run: + +```bash +uv run pytest tests/e2e/test_smoke_real_sandbox_posture.py::test_smoke_contract_declares_synthetic_rl_tokens_and_logprobs -q +``` + +Expected: FAIL because the constants do not exist. + +- [ ] **Step 3: Add RL metadata constants** + +In `tests/fixtures/smoke_components/smoke_base/constants.py`, add: + +```python +EXPECTED_SMOKE_TOKEN_IDS: list[int] = [101, 202, 303] +EXPECTED_SMOKE_LOGPROBS: list[float] = [-0.10, -0.20, -0.30] +EXPECTED_SUBAGENT_INTERNAL_MARKER = "SMOKE_CHILD_INTERNAL_THINKING" +EXPECTED_PARENT_VISIBLE_CHILD_RESULT = "SMOKE_CHILD_RESULT_PARENT_VISIBLE" +``` + +- [ ] **Step 4: Emit synthetic token/logprob metadata from a smoke leaf** + +In `tests/fixtures/smoke_components/smoke_base/leaf_base.py`, make at least one deterministic leaf stream a context/tool/action event whose persisted token metadata includes: + +```python +{ + "token_ids": [101, 202, 303], + "logprobs": [ + {"token_id": 101, "token": "smoke", "logprob": -0.10}, + {"token_id": 202, "token": "handoff", "logprob": -0.20}, + {"token_id": 303, "token": "ok", "logprob": -0.30}, + ], +} +``` + +Use the same event path that real workers/tool calls use so `RlEpisodeReadService` and `RlProjectionService` see the metadata. Do not add a special-case RL test table writer. + +Also emit deterministic text markers through real context events: + +- a child-internal action text containing `EXPECTED_SUBAGENT_INTERNAL_MARKER` +- a parent-visible tool result / child result text containing `EXPECTED_PARENT_VISIBLE_CHILD_RESULT` + +The smoke RL assertion should verify the parent actor span excludes the child-internal marker while the joint timeline still contains it, and that the parent actor span includes the parent-visible child result. This preserves the formalism we wanted: child internals exist in the joint episode tree, but the parent learns from the spawn action and returned result, not from the child's private rollout. + +- [ ] **Step 5: Add host-side RL projection assertion** + +In `tests/e2e/_asserts.py`, add: + +```python +def _assert_smoke_rl_view_shape(sample_id: UUID) -> None: + from ergon_core.core.persistence.shared.db import get_session + from ergon_core.core.views.rl.episode import RlEpisodeReadService + from ergon_core.core.views.rl.projections import RlProjectionService + from tests.fixtures.smoke_components.smoke_base.constants import ( + EXPECTED_PARENT_VISIBLE_CHILD_RESULT, + EXPECTED_SMOKE_LOGPROBS, + EXPECTED_SMOKE_TOKEN_IDS, + EXPECTED_SUBAGENT_INTERNAL_MARKER, + EXPECTED_SUBTASK_SLUGS, + ) + + with get_session() as session: + episode = RlEpisodeReadService(session).get_episode(sample_id) + + assert episode.sample_id == sample_id + assert episode.experiment_id is not None + assert episode.environment_id is not None + assert episode.sample_key is not None + assert episode.normalized_reward in {0.0, 1.0} + assert len(episode.root_tasks) == 1 + + root = episode.root_tasks[0] + assert root.parent_task_id is None + assert root.level == 0 + assert root.actor is not None + assert root.actor.actor_slug + assert root.actor.parent_actor_slug is None + assert sorted(child.task_slug for child in root.children) == sorted(EXPECTED_SUBTASK_SLUGS) + + producer = next( + child for child in root.children if child.task_slug == EXPECTED_RESOURCE_HANDOFF.producer_slug + ) + consumer = next( + child for child in root.children if child.task_slug == EXPECTED_RESOURCE_HANDOFF.consumer_slug + ) + assert producer.actor is not None + assert consumer.actor is not None + assert consumer.actor.parent_actor_slug == root.actor.actor_slug + assert consumer.parent_task_id == root.task_id + assert all(attempt.attempt_number >= 1 for attempt in consumer.attempts) + + projector = RlProjectionService() + timeline = projector.get_joint_timeline(episode) + assert timeline.records, "expected joint RL timeline records" + assert all(record.actor.actor_slug for record in timeline.records) + assert all(record.task_id is not None for record in timeline.records) + assert all(record.task_attempt_id is not None for record in timeline.records) + assert timeline.records == sorted( + timeline.records, + key=lambda record: ( + record.created_at is None, + record.created_at, + record.sequence, + str(record.task_id), + str(record.task_attempt_id), + ), + ) + + actor_spans = list(projector.iter_actor_spans(episode, group_by="actor_slug")) + assert actor_spans, "expected actor-grouped RL spans" + root_span = next(span for span in actor_spans if span.key == root.actor.actor_slug) + assert all(record.actor.actor_slug == root.actor.actor_slug for record in root_span.records) + assert EXPECTED_SUBAGENT_INTERNAL_MARKER not in [record.text for record in root_span.records] + assert any(EXPECTED_PARENT_VISIBLE_CHILD_RESULT in record.text for record in root_span.records) + assert any(EXPECTED_SUBAGENT_INTERNAL_MARKER in record.text for record in timeline.records) + + trajectories = list(projector.iter_task_attempt_trajectories(episode)) + assert trajectories, "expected task-attempt trajectories" + assert {trajectory.task_slug for trajectory in trajectories} >= { + EXPECTED_RESOURCE_HANDOFF.producer_slug, + EXPECTED_RESOURCE_HANDOFF.consumer_slug, + } + assert all(trajectory.task_attempt_id is not None for trajectory in trajectories) + + action_records = [record for record in timeline.records if record.step_kind == "action"] + token_ids = [ + token_id + for record in action_records + if record.token_metadata is not None + for token_id in (record.token_metadata.token_ids or []) + ] + logprobs = [ + item.logprob + for record in action_records + if record.token_metadata is not None and record.token_metadata.logprobs is not None + for item in record.token_metadata.logprobs + ] + assert EXPECTED_SMOKE_TOKEN_IDS == token_ids[: len(EXPECTED_SMOKE_TOKEN_IDS)] + assert EXPECTED_SMOKE_LOGPROBS == logprobs[: len(EXPECTED_SMOKE_LOGPROBS)] +``` + +This assertion should use the new PR-stack RL APIs directly: + +- `RlEpisodeReadService.get_episode(sample_id)` +- `RlProjectionService.iter_actor_spans(...)` +- `RlProjectionService.iter_task_attempt_trajectories(...)` +- `RlProjectionService.get_joint_timeline(...)` + +Do not assert by scraping raw telemetry tables. If importing production services directly in E2E assertions becomes too heavy, add a thin helper in `ergon_core/ergon_core/test_support/e2e_read_helpers.py` that returns these service DTOs unchanged. + +- [ ] **Step 6: Add trainer-facing training-record shape assertion** + +In `tests/e2e/_asserts.py`, add a second assertion for the trainer-facing shape produced from the RL view: + +```python +def _assert_smoke_training_record_shape(sample_id: UUID) -> None: + from ergon_core.core.persistence.shared.db import get_session + from ergon_core.core.rl.rollout_service import _logprobs_for_kind, _token_ids_for_kind + from ergon_core.core.views.rl.episode import RlEpisodeReadService + from ergon_core.core.views.rl.projections import RlProjectionService + from tests.fixtures.smoke_components.smoke_base.constants import ( + EXPECTED_SMOKE_LOGPROBS, + EXPECTED_SMOKE_TOKEN_IDS, + ) + + with get_session() as session: + episode = RlEpisodeReadService(session).get_episode(sample_id) + + spans = list(RlProjectionService().iter_actor_spans(episode, group_by="actor_slug")) + records = [] + for span in spans: + prompt_ids = _token_ids_for_kind(span.records, "observation") + completion_ids = _token_ids_for_kind(span.records, "action") + logprobs = _logprobs_for_kind(span.records, "action") + if prompt_ids or completion_ids: + first = span.records[0] + records.append( + { + "sample_id": sample_id, + "actor_slug": first.actor.actor_slug, + "base_worker_slug": first.actor.base_worker_slug, + "parent_actor_slug": first.actor.parent_actor_slug, + "task_id": first.task_id, + "task_attempt_id": first.task_attempt_id, + "prompt_ids": prompt_ids, + "completion_ids": completion_ids, + "logprobs": logprobs, + "reward": episode.normalized_reward or 0.0, + } + ) + + assert records, "expected at least one trainer-facing projected record" + record_with_tokens = next(record for record in records if record["completion_ids"]) + assert record_with_tokens["sample_id"] == sample_id + assert record_with_tokens["actor_slug"] + assert record_with_tokens["task_id"] is not None + assert record_with_tokens["task_attempt_id"] is not None + assert record_with_tokens["reward"] in {0.0, 1.0} + assert record_with_tokens["completion_ids"][:3] == EXPECTED_SMOKE_TOKEN_IDS + assert record_with_tokens["logprobs"][:3] == EXPECTED_SMOKE_LOGPROBS +``` + +This mirrors the current `RolloutService._build_training_records(...)` behavior without needing to create a rollout batch inside smoke. It specifically asserts the adapter-facing shape: + +- sample id +- actor identity +- prompt ids from observation steps +- completion ids/logprobs from action steps +- sample-level normalized reward +- task id and task attempt id + +- [ ] **Step 7: Tighten root and intermediate evaluation assertions** + +In `tests/e2e/_asserts.py`, extend `_assert_temporal_ordering(sample_id)` or add a new helper. Keep this host-side; the only new runtime evaluator is the single dynamic-subtask evaluator from Task 6. + +```python +def _assert_evaluations_after_required_outputs(sample_id: UUID) -> None: + root_execution, evaluations = list_root_execution_and_evaluations(sample_id) + assert root_execution is not None + resources = list_named_resources( + sample_id, + prefix=HANDOFF_RESOURCE_NAME.removesuffix(".json"), + suffix=".json", + ) + assert resources, "handoff resource missing before evaluation" + for evaluation in evaluations: + assert evaluation.created_at >= resources[0].created_at +``` + +Also assert the intermediary dynamic evaluation exists on happy path: + +```python +def _assert_intermediate_evaluation_ordering(sample_id: UUID) -> None: + snapshot = require_run_snapshot(sample_id) + consumer = next( + task + for task in snapshot.tasks.values() + if task.name == EXPECTED_RESOURCE_HANDOFF.consumer_slug + ) + evaluation = snapshot.evaluations_by_task.get(consumer.id) + assert evaluation is not None + handoffs = list_named_resources( + sample_id, + prefix=HANDOFF_RESOURCE_NAME.removesuffix(".json"), + suffix=".json", + ) + assert handoffs + assert evaluation.normalized_score == 1.0 + assert evaluation.created_at >= handoffs[0].created_at +``` + +The invariant is structural: root and intermediate evaluator rows must not appear before the resources/messages they claim to evaluate. Sad-path smoke should assert there is no successful intermediate evaluation for a blocked or unstarted consumer task. + +- [ ] **Step 8: Add deterministic graph/team mutation lane** + +In `tests/fixtures/smoke_components/smoke_base/worker_base.py`, add one controlled mutation during parent execution: + +```python +# after planning roots, before awaiting children +# Add a transient worker binding or metadata annotation, then remove/supersede it. +await context.record_team_event( + event_type="worker_binding_added", + worker_slug="smoke-transient-reviewer", + reason="smoke mutation lane", +) +await context.record_team_event( + event_type="worker_binding_removed", + worker_slug="smoke-transient-reviewer", + reason="smoke mutation lane complete", +) +``` + +If no worker/team event API exists, use the closest existing graph mutation API and rename this lane to a graph metadata mutation. The goal is deterministic coverage of add/remove lifecycle visibility, not a new production feature. + +- [ ] **Step 9: Assert mutation visibility** + +In `tests/e2e/_asserts.py`, add: + +```python +def _assert_smoke_dynamic_mutation_events(sample_id: UUID) -> None: + stream = read_sample_runtime_event_stream(sample_id) + event_types = [event.event_type for event in stream.events] + assert "worker_binding_added" in event_types + assert "worker_binding_removed" in event_types + assert event_types.index("worker_binding_added") < event_types.index("worker_binding_removed") +``` + +If the implemented lane uses graph metadata events, assert those concrete event names instead. + +- [ ] **Step 10: Wire new assertions into happy and sad smoke tests** + +In each E2E smoke file: + +```python +_assert_resource_handoff_and_coordination_thread(rid) +_assert_intermediate_handoff_evaluation(rid) +_assert_smoke_rl_view_shape(rid) +_assert_smoke_training_record_shape(rid) +_assert_evaluations_after_required_outputs(rid) +_assert_intermediate_evaluation_ordering(rid) +_assert_smoke_dynamic_mutation_events(rid) +``` + +For sad path, use sad-specific variants where a blocked task should not have reply/logprob/evaluation artifacts. + +- [ ] **Step 11: Verify** + +Run: + +```bash +uv run pytest tests/e2e/test_smoke_real_sandbox_posture.py -q +ergon test smoke +``` + +Expected: PASS. Failures here are valuable: they indicate the smoke scenario is now catching real resource, message, RL metadata, ordering, or mutation bugs. + +- [ ] **Step 12: Commit** + +```bash +git add tests/fixtures/smoke_components/smoke_base/constants.py tests/fixtures/smoke_components/smoke_base/leaf_base.py tests/fixtures/smoke_components/smoke_base/worker_base.py tests/e2e/_asserts.py ergon_core/ergon_core/test_support/e2e_read_helpers.py tests/e2e/test_smoke_real_sandbox_posture.py tests/e2e/test_researchrubrics_smoke.py tests/e2e/test_minif2f_smoke.py tests/e2e/test_swebench_smoke.py +git commit -m "test: tighten smoke runtime semantic assertions" +``` + +--- + +## Task 9: Update Comments, Docs, And Test Naming Around Smoke Posture + +**Files:** +- Modify: `tests/fixtures/smoke_components/sandbox.py` +- Modify: `tests/e2e/test_researchrubrics_smoke.py` +- Modify: `tests/e2e/test_minif2f_smoke.py` +- Modify: `tests/e2e/test_swebench_smoke.py` +- Modify if present: CLI docs mentioning `ergon test smoke` + +- [ ] **Step 1: Update stale comments** + +Change misleading comments that describe fake sandbox use as intentional canonical smoke behavior. `tests/fixtures/smoke_components/sandbox.py` should say it is legacy/local-unit support only if any references remain. + +Use wording like: + +```python +"""Legacy local sandbox implementation. + +Canonical `ergon test smoke` must use real benchmark sandbox classes. +This module exists only for narrow unit tests that need an in-process +SandboxRuntime without provider access. +""" +``` + +- [ ] **Step 2: Keep E2E smoke docstrings honest** + +For each E2E smoke file, use: + +```python +""" canonical happy/sad smoke experiment group using real benchmark sandboxes.""" +``` + +- [ ] **Step 3: Run comment/docs-adjacent tests** + +Run: + +```bash +uv run pytest tests/e2e/test_smoke_real_sandbox_posture.py -q +``` + +Expected: PASS. + +- [ ] **Step 4: Commit** + +```bash +git add tests/fixtures/smoke_components/sandbox.py tests/e2e/test_researchrubrics_smoke.py tests/e2e/test_minif2f_smoke.py tests/e2e/test_swebench_smoke.py +git commit -m "docs: clarify smoke sandbox posture" +``` + +--- + +## Task 10: Run The Real Smoke Suite Locally + +**Files:** +- No code changes expected. + +- [ ] **Step 1: Start the dev stack** + +Run: + +```bash +ergon start +``` + +Expected: Dashboard, API, Inngest, and Postgres are healthy. + +- [ ] **Step 2: Run the new posture contract** + +Run: + +```bash +uv run pytest tests/e2e/test_smoke_real_sandbox_posture.py -q +``` + +Expected: PASS. + +- [ ] **Step 3: Run canonical smoke** + +Run: + +```bash +ergon test smoke +``` + +Expected: PASS for researchrubrics, minif2f, and swebench smoke tests. + +Important expected behavior change: this now consumes real sandbox provider resources and can fail for real provider/image/runtime problems. That is the point of this PR. + +- [ ] **Step 4: If provider credentials or quota are missing** + +Do not weaken `ergon test smoke`. If the tighter smoke posture exposes a real runtime or infrastructure bug, keep the stricter assertion and mark the affected smoke test `xfail` with the exact bug reason. Do not make the implementation softer just to satisfy the test. + +Known expected failures discovered while executing this plan: + +- The real benchmark sandbox classes use private E2B template names (`ergon-research-v1`, `ergon-minif2f-v1`, `ergon-swebench-v1`) that may not exist in the executing account. +- The public object-bound sandbox path does not currently emit the command WAL rows that the old smoke-only `InstrumentedSandbox` path emitted. +- The current E2B detach path calls an SDK `close()` method that is not present on the locally installed `AsyncSandbox` object. + +These are intentionally represented as `pytest.mark.xfail` on the live E2E smoke drivers. When the underlying bugs/configuration are fixed, the xfails should become xpasses and then be removed. + +- [ ] **Step 5: Commit any final fixes** + +```bash +git status --short +git add +git commit -m "test: verify real sandbox smoke suite" +``` + +--- + +## Task 11: Push Top-Of-Stack PR + +**Files:** +- No code changes expected. + +- [ ] **Step 1: Create the branch from the current stack tip** + +Run from current PR stack tip: + +```bash +git switch codex/episode-sample-pr13-rl-views-stack +git pull --ff-only +git switch -c codex/episode-sample-pr14-real-sandbox-smoke +``` + +- [ ] **Step 2: Rebase if the lower stack moves** + +If PR #132 changes before this PR is opened: + +```bash +git fetch origin +git rebase origin/codex/episode-sample-pr13-rl-views-stack +``` + +- [ ] **Step 3: Push** + +```bash +git push -u origin codex/episode-sample-pr14-real-sandbox-smoke +``` + +- [ ] **Step 4: Open PR** + +```bash +gh pr create \ + --base codex/episode-sample-pr13-rl-views-stack \ + --head codex/episode-sample-pr14-real-sandbox-smoke \ + --title "Run canonical smoke tests with real benchmark sandboxes" \ + --body "## Summary +- switches canonical smoke fixtures from fake SmokePublicSandbox to each environment's real sandbox class +- adds posture tests/guards preventing SmokePublicSandbox or TestSandbox from entering canonical smoke task snapshots +- improves smoke root/child task names and descriptions so the real sandbox run catches more realistic task rendering, command, artifact, and dependency bugs +- seeds realistic smoke documents and adds resource-handoff, coordination-thread, RL logprob, root/intermediate evaluation, and dynamic mutation assertions +- adds exactly one dynamic-subtask evaluator for intermediary terminal-state coverage +- exercises production builtin toolkits from deterministic smoke workers for free integration coverage of command/file/report/proof/repo tools +- updates smoke comments/docs so provider coverage is explicit + +## Verification +- uv run pytest tests/unit/smoke_base/test_smoke_real_sandbox_posture.py -q +- uv run ruff check +- ERGON_DATABASE_URL=postgresql://ergon:ergon_dev@localhost:5433/ergon INNGEST_API_BASE_URL=http://localhost:8289 uv run pytest tests/e2e/test_researchrubrics_smoke.py::test_smoke_experiment_group tests/e2e/test_minif2f_smoke.py::test_smoke_experiment_group tests/e2e/test_swebench_smoke.py::test_smoke_experiment_group --collect-only -q +- ergon test smoke (currently expected to report xfailed live smoke tests until the real sandbox bugs above are fixed)" +``` + +- [ ] **Step 5: Watch CI** + +Run: + +```bash +gh pr checks --watch --fail-fast +``` + +Expected: CI passes. If CI provider resources are missing, treat that as infrastructure/configuration truth, not a reason to silently reintroduce fake sandboxes. + +--- + +## Self-Review + +- Spec coverage: The plan changes canonical smoke posture from fake local sandbox to real environment sandbox classes, improves smoke task names/descriptions to cover realistic task rendering and sandbox-runtime behaviors, exercises production builtin toolkits from deterministic smoke workers, adds seeded document/resource/message semantics, asserts RL logprobs and root/intermediate evaluation ordering, adds exactly one dynamic-subtask evaluator for intermediary terminal-state coverage, adds deterministic mutation coverage, guards against fake/test sandboxes, updates comments, and verifies with both a fast contract test and `ergon test smoke`. +- Placeholder scan: No TBD/TODO placeholders remain. Each task has files, code snippets, commands, and expected outcomes. +- Type consistency: Test imports use existing environment class names; sandbox type strings match existing `_type` discriminator format observed in task snapshots; branch base matches the current top-of-stack PR branch. + +--- + +## Follow-Up Fix Plan For Current Live Smoke XFails + +The real-sandbox posture PR intentionally xfails the live smoke tests when tightening the suite exposes platform bugs. Do not soften the smoke assertions to make these pass. Fix the ownership boundaries below, then remove the relevant xfail reason only after the targeted unit/integration coverage is green. + +### Fix 1: Sandbox Template Provisioning Belongs To Experiment Preflight + +**Current symptom:** real E2B-backed smoke runs can fail before worker logic executes because private or locally built sandbox templates are unavailable. + +**Ownership decision:** smoke tests should not build templates, and pure experiment persistence should not perform network/build side effects. The right owner is the experiment submission/preflight path immediately before experiment materialization/execution. That boundary already knows which environments/sandboxes the run will use and can fail early with an actionable setup error before samples are launched. + +**Architecture:** +- Add a template requirement/resolution service that reads environment catalog metadata and the local sandbox template registry. +- Run that resolver during experiment submission/preflight. +- Persist the resolved template reference into task/runtime snapshots. +- Keep worker/tool authors unaware of RL reconstruction or sandbox template provisioning concerns. + +**Files to inspect/modify:** +- `ergon_builtins/ergon_builtins/environments/catalog.py` +- `ergon_cli/domains/environments/templates.py` +- `ergon_builtins/ergon_builtins/benchmarks/minif2f/sandbox_template/utils.py` +- `ergon_builtins/ergon_builtins/benchmarks/minif2f/sandbox.py` +- experiment submission/preflight entrypoints in `ergon_core/ergon_core/core/jobs/` and CLI submit code + +**Implementation steps:** +- [ ] Write a unit test proving a known environment exposes a required sandbox template requirement. +- [ ] Write a unit test proving the resolver prefers a pinned local registry `template_id` over a friendly template name. +- [ ] Write a unit test proving missing templates raise a typed, actionable preflight error before samples are submitted. +- [ ] Implement the resolver with no smoke-specific logic. +- [ ] Wire the resolver into experiment submission/preflight, not into the smoke fixture. +- [ ] Update the real smoke xfail reason to remove template provisioning only after a real sandbox smoke can pass that phase locally/CI with provisioned templates. + +**Recommendation:** make auto-build opt-in, not implicit. Default behavior should be a fast preflight failure with a command or setup instruction. Template building can be expensive, require credentials, and should not surprise users during normal experiment persistence. + +### Fix 2: Public Sandbox Runtime Calls Must Emit Command WAL + +**Current symptom:** deterministic smoke workers use the public `Task.sandbox` API and production builtin toolkits, but those calls do not necessarily emit the same command/file WAL rows as the older instrumented sandbox path. + +**Ownership decision:** instrumentation belongs at the sandbox runtime boundary, not in smoke workers or individual toolkits. The public sandbox API should be observable by default when attached to a running sample/task context. + +**Architecture:** +- Add an `InstrumentedSandboxRuntime` wrapper around the `SandboxRuntime` protocol. +- Wrap real runtimes when task/evaluator jobs attach or provision a sandbox. +- Emit sandbox WAL entries for `run_command`, `write_file`, `read_file`, and `list_files`. +- Delegate lifecycle methods such as `close()` and `close_local()` to the wrapped runtime. + +**Files to inspect/modify:** +- `ergon_core/ergon_core/api/sandbox/sandbox.py` +- `ergon_core/ergon_core/api/sandbox/runtime.py` +- `ergon_builtins/ergon_builtins/sandbox/e2b_runtime.py` +- `ergon_core/ergon_core/core/infrastructure/sandbox/instrumentation.py` +- task worker/evaluator job files under `ergon_core/ergon_core/core/jobs/task/` +- smoke assertions in `tests/e2e/_asserts.py` + +**Implementation steps:** +- [ ] Write a unit test with a fake `SandboxRuntime` proving `run_command` delegates and emits exactly one command WAL event with sample/task/execution/sandbox identity. +- [ ] Write equivalent unit tests for `write_file`, `read_file`, and `list_files`. +- [ ] Implement `InstrumentedSandboxRuntime`. +- [ ] Wire runtime instrumentation into worker execution and evaluator execution attachment points. +- [ ] Add an integration or smoke assertion that builtin toolkit activity produces sandbox command/file WAL rows. +- [ ] Remove only the WAL-related xfail reason after the live smoke assertion passes without synthetic event shims. + +**Recommendation:** instrument the public runtime layer even if the old `InstrumentedSandbox` remains for compatibility. That makes object-bound `task.sandbox` calls, builtin toolkit calls, and future runtime implementations share one observability contract. + +### Fix 3: E2B Local Detach Has No SDK Close Method + +**Current symptom:** the current E2B runtime contains a local detach path that calls `self._sandbox.close()`, but the installed E2B SDK exposes no `close`, `disconnect`, or `shutdown` method on `AsyncSandbox`. + +**SDK fact checked locally:** +- `AsyncSandbox.kill()` exists and terminates the remote sandbox. +- `AsyncSandbox.pause()` exists and changes remote lifecycle. +- `AsyncSandbox.close()` does not exist. +- There is no SDK method that means "drop this local Python handle while leaving the remote sandbox running." + +**Ownership decision:** model local detach honestly in the runtime adapter. For E2B, detach is a no-op because the SDK object is a lightweight API client/handle, not a local stream that needs closing. + +**Files to inspect/modify:** +- `ergon_builtins/ergon_builtins/sandbox/e2b_runtime.py` +- unit tests for E2B runtime lifecycle behavior + +**Implementation steps:** +- [ ] Write a unit test proving `E2BSandboxRuntime.close()` calls `kill()`. +- [ ] Write a unit test proving `E2BSandboxRuntime.close_local()` does not call `kill()` or `pause()`. +- [ ] Replace `close_local()` with a documented no-op explaining the SDK has no local-only close primitive. +- [ ] Run the E2B runtime lifecycle tests. +- [ ] Remove only the detach-related xfail reason after the live smoke no longer fails on missing `close()`. + +**Recommendation:** do not use `pause()` as a substitute for detach. Pausing mutates remote sandbox lifecycle and is not equivalent to dropping the local handle. diff --git a/ergon-dashboard/tests/e2e/_shared/expected.ts b/ergon-dashboard/tests/e2e/_shared/expected.ts index 5b0344e32..811288547 100644 --- a/ergon-dashboard/tests/e2e/_shared/expected.ts +++ b/ergon-dashboard/tests/e2e/_shared/expected.ts @@ -7,15 +7,18 @@ */ export const EXPECTED_SUBTASK_SLUGS = [ - "d_root", - "d_left", - "d_right", - "d_join", - "l_1", - "l_2", - "l_3", - "s_a", - "s_b", + "source-review", + "handoff-verify", + "primary-artifact", + "artifact-summary", + "metadata-review", + "environment-probe", + "metadata-validate", + "evidence-artifact", + "completion-marker", ] as const; -export const EXPECTED_NESTED_SUBTASK_SLUGS = ["l_2_a", "l_2_b"] as const; +export const EXPECTED_NESTED_SUBTASK_SLUGS = [ + "nested-input-review", + "nested-verification", +] as const; diff --git a/ergon-dashboard/tests/e2e/_shared/smoke.ts b/ergon-dashboard/tests/e2e/_shared/smoke.ts index 21bf9b296..f518f3d47 100644 --- a/ergon-dashboard/tests/e2e/_shared/smoke.ts +++ b/ergon-dashboard/tests/e2e/_shared/smoke.ts @@ -79,7 +79,7 @@ async function selectRenderedGraphTask( evaluatedTaskIds: Set, ): Promise { const candidates = [ - ...state.graph_nodes.filter((node) => node.level > 0 && node.task_slug === "d_root"), + ...state.graph_nodes.filter((node) => node.level > 0 && node.task_slug === "source-review"), ...state.graph_nodes.filter((node) => node.level > 0 && evaluatedTaskIds.has(node.id)), ...state.graph_nodes.filter((node) => node.level > 0), ]; @@ -272,7 +272,7 @@ export function defineSmokeSpec(cfg: SmokeSpecConfig): void { return; } - // Canonical sad path: l_2 fails, l_3 blocks, independent leaves complete. + // Canonical sad path: environment-probe fails, metadata-validate blocks. expect(state.status).toBe("failed"); expect(state.resource_count).toBeGreaterThanOrEqual(15); expect(state.executions.length).toBe(state.execution_count); @@ -282,11 +282,11 @@ export function defineSmokeSpec(cfg: SmokeSpecConfig): void { const statusBySlug = new Map( state.graph_nodes.filter((n) => n.level > 0).map((n) => [n.task_slug, n.status]), ); - for (const slug of EXPECTED_SUBTASK_SLUGS.filter((s) => !["l_2", "l_3"].includes(s))) { + for (const slug of EXPECTED_SUBTASK_SLUGS.filter((s) => !["environment-probe", "metadata-validate"].includes(s))) { expect(statusBySlug.get(slug)).toBe("completed"); } - expect(statusBySlug.get("l_2")).toBe("failed"); - expect(statusBySlug.get("l_3")).toBe("blocked"); + expect(statusBySlug.get("environment-probe")).toBe("failed"); + expect(statusBySlug.get("metadata-validate")).toBe("blocked"); await page.goto(`/samples/${sample_id}`); await assertRunWorkspace(page, state, sample_id); diff --git a/tests/e2e/_asserts.py b/tests/e2e/_asserts.py index a4ea672c5..79bb4b9a4 100644 --- a/tests/e2e/_asserts.py +++ b/tests/e2e/_asserts.py @@ -23,6 +23,8 @@ from uuid import UUID import httpx +from ergon_core.core.views.rl.episode import RlEpisodeReadService +from ergon_core.core.views.rl.projections import RlProjectionService from ergon_core.core.views.samples.models import SampleTaskDto from ergon_core.test_support.e2e_read_helpers import ( ObservedSampleRuntimeEvent, @@ -37,7 +39,26 @@ read_sample_runtime_event_stream, read_resource_bytes, ) -from tests.fixtures.smoke_components.smoke_base.constants import EXPECTED_SUBTASK_SLUGS +from tests.fixtures.smoke_components.smoke_base.constants import ( + ARTIFACT_SUMMARY_SLUG, + ENV_PROBE_SLUG, + EXPECTED_RESOURCE_HANDOFF, + EXPECTED_SMOKE_LOGPROBS, + EXPECTED_SMOKE_TOKEN_IDS, + EXPECTED_SUBTASK_SLUGS, + EXPECTED_SUBAGENT_INTERNAL_MARKER, + EXPECTED_PARENT_VISIBLE_CHILD_RESULT, + HANDOFF_RESOURCE_NAME, + HANDOFF_VERIFY_SLUG, + METADATA_REVIEW_SLUG, + METADATA_VALIDATE_SLUG, + NESTED_INSPECT_SLUG, + NESTED_LINE_SLUGS, + NESTED_VERIFY_SLUG, + PRIMARY_ARTIFACT_SLUG, + SEEDED_SOURCE_DOC_NAME, + SOURCE_REVIEW_SLUG, +) from tests.e2e._read_contracts import require_run_snapshot @@ -45,7 +66,6 @@ BLOCKED = "blocked" COMPLETED = "completed" FAILED = "failed" -NESTED_LINE_SLUGS = ("l_2_a", "l_2_b") SMOKE_PARENT_TURN_COUNT = 3 SMOKE_RECURSIVE_TURN_COUNT = 3 SMOKE_LEAF_TURN_COUNT = 2 @@ -73,9 +93,9 @@ def _assert_sample_graph(sample_id: UUID) -> None: EXPECTED_SUBTASK_SLUGS, ) assert sorted(task.name for task in tasks if task.level == 2) == sorted(NESTED_LINE_SLUGS) - assert by_slug["l_2"].is_leaf is False - assert by_slug["l_2_a"].parent_id == by_slug["l_2"].id - assert by_slug["l_2_b"].parent_id == by_slug["l_2"].id + assert by_slug[ENV_PROBE_SLUG].is_leaf is False + assert by_slug[NESTED_INSPECT_SLUG].parent_id == by_slug[ENV_PROBE_SLUG].id + assert by_slug[NESTED_VERIFY_SLUG].parent_id == by_slug[ENV_PROBE_SLUG].id non_completed = [(task.name, task.status) for task in tasks if task.status != COMPLETED] assert not non_completed, f"non-completed nodes: {non_completed}" @@ -92,20 +112,20 @@ def _assert_dag_edges(leaves: list[SampleTaskDto]) -> None: if parent_id in by_id } expected_pairs = { - ("d_root", "d_left"), - ("d_root", "d_right"), - ("d_left", "d_join"), - ("d_right", "d_join"), - ("l_1", "l_2"), - ("l_2", "l_3"), - ("l_2_a", "l_2_b"), + (SOURCE_REVIEW_SLUG, HANDOFF_VERIFY_SLUG), + (SOURCE_REVIEW_SLUG, PRIMARY_ARTIFACT_SLUG), + (HANDOFF_VERIFY_SLUG, ARTIFACT_SUMMARY_SLUG), + (PRIMARY_ARTIFACT_SLUG, ARTIFACT_SUMMARY_SLUG), + (METADATA_REVIEW_SLUG, ENV_PROBE_SLUG), + (ENV_PROBE_SLUG, METADATA_VALIDATE_SLUG), + (NESTED_INSPECT_SLUG, NESTED_VERIFY_SLUG), } missing = expected_pairs - actual_pairs assert not missing, f"missing DAG edges: {missing}" def _assert_sample_resources(sample_id: UUID) -> None: - """Exactly 20 task resources: 10 benchmark artifacts + 10 probe_*.json.""" + """Expected task resources include artifacts, probes, toolkit probes, and handoff files.""" snapshot = require_run_snapshot(sample_id) resources = [ resource @@ -122,13 +142,25 @@ def _assert_sample_resources(sample_id: UUID) -> None: assert not worker_outputs, ( "worker final assistant messages must stay on executions, not resources" ) - assert len(resources) == 20, ( - f"expected 20 task artifact resources (10 outputs + 10 probes), got {len(resources)}" + toolkit_probes = [ + resource + for resource in resources + if resource.name.startswith("toolkit_probe_") and resource.name.endswith(".json") + ] + assert len(toolkit_probes) == 10, ( + f"expected 10 toolkit_probe_*.json resources, got {len(toolkit_probes)}" + ) + names = {resource.name for resource in resources} + assert SEEDED_SOURCE_DOC_NAME in names + assert HANDOFF_RESOURCE_NAME in names + assert len(resources) == 32, ( + f"expected 32 task artifact resources " + f"(10 outputs + 10 probes + 10 toolkit probes + source/handoff), got {len(resources)}" ) def _assert_run_turn_counts(sample_id: UUID) -> None: - """Parent + recursive ``l_2`` + artifact leaves emit fixed chunk counts. + """Parent + recursive ``environment-probe`` + artifact leaves emit fixed chunk counts. Each smoke context chunk contains one assistant text part, so persistence emits exactly one ``SampleContextEvent`` per chunk. @@ -196,6 +228,66 @@ def _assert_run_evaluation(sample_id: UUID) -> None: assert criterion.contribution >= 0 +def _assert_handoff_task_evaluation(sample_id: UUID) -> None: + deadline = time.monotonic() + 30 + evaluation = None + consumer = None + while time.monotonic() < deadline: + snapshot = require_run_snapshot(sample_id) + consumer = next( + (task for task in snapshot.tasks.values() if task.name == HANDOFF_VERIFY_SLUG), + None, + ) + if consumer is not None: + evaluation = snapshot.evaluations_by_task.get(consumer.id) + if evaluation is not None: + break + time.sleep(2) + + assert consumer is not None, f"expected {HANDOFF_VERIFY_SLUG} task in snapshot" + assert evaluation is not None, f"expected evaluation for {HANDOFF_VERIFY_SLUG}" + assert evaluation.normalized_score == 1.0 + criterion_slugs = {criterion.criterion_slug for criterion in evaluation.criterion_results} + assert criterion_slugs == {"smoke-resource-handoff"} + + +def _assert_rl_episode_view(sample_id: UUID) -> None: + episode = RlEpisodeReadService().get_episode(sample_id) + assert episode.sample_id == sample_id + assert episode.normalized_reward == 1.0 + assert len(episode.root_tasks) == 1 + + tasks = _walk_rl_tasks(episode.root_tasks) + by_slug = {task.task_slug: task for task in tasks} + assert set(EXPECTED_SUBTASK_SLUGS) <= set(by_slug) + assert by_slug[ENV_PROBE_SLUG].children + assert {child.task_slug for child in by_slug[ENV_PROBE_SLUG].children} == set(NESTED_LINE_SLUGS) + assert by_slug[HANDOFF_VERIFY_SLUG].actor is not None + assert ( + by_slug[HANDOFF_VERIFY_SLUG].actor.parent_task_id + == by_slug[SOURCE_REVIEW_SLUG].parent_task_id + ) + + projection = RlProjectionService() + trajectories = list(projection.iter_task_attempt_trajectories(episode)) + assert {trajectory.task_slug for trajectory in trajectories} >= set(EXPECTED_SUBTASK_SLUGS) + joint_timeline = projection.get_joint_timeline(episode) + token_steps = [step for step in joint_timeline.records if step.token_metadata is not None] + assert token_steps, "expected RL joint timeline to expose synthetic logprobs" + first = token_steps[0].token_metadata + assert first is not None + assert first.token_ids == EXPECTED_SMOKE_TOKEN_IDS + assert [item.logprob for item in first.logprobs or []] == EXPECTED_SMOKE_LOGPROBS + + +def _walk_rl_tasks(tasks) -> list: + result = [] + for task in tasks: + result.append(task) + result.extend(_walk_rl_tasks(task.children)) + return result + + # ============================================================================= # Observability helpers (run-level) # ============================================================================= @@ -205,8 +297,8 @@ def _assert_sandbox_command_wal(sample_id: UUID) -> None: """Bash commands land as WAL rows via ``PostgresSandboxEventSink``.""" entries = list_sandbox_command_wal(sample_id) probes = [e for e in entries if "wc" in e.command or "probe" in e.command] - # Canonical sad-path smokes block l_3 before it starts, so the eight - # executed leaves should emit probe commands while l_3 emits none. + # Canonical sad-path smokes block metadata-validate before it starts, so + # executed leaves should emit probe commands while the blocked node emits none. assert len(probes) >= 8, f"expected ≥8 probe WAL entries, got {len(probes)}" @@ -266,6 +358,40 @@ def _assert_blob_roundtrip(sample_id: UUID) -> None: assert "exit_code" in parsed, f"probe JSON missing exit_code: {parsed!r}" +def _assert_toolkit_probe_resources( + sample_id: UUID, + *, + expected_toolkit_suffix: str, +) -> None: + resources = _require_named_resources( + sample_id, + prefix="toolkit_probe_", + suffix=".json", + expected_count=10, + ) + for resource in resources: + payload = json.loads(read_resource_bytes(resource)) + assert payload["ok"] is True, f"{resource.name} toolkit probe failed: {payload!r}" + assert payload["toolkit"].endswith(expected_toolkit_suffix) + assert payload["synthetic_token_ids"] == EXPECTED_SMOKE_TOKEN_IDS + assert payload["synthetic_logprobs"] == EXPECTED_SMOKE_LOGPROBS + assert payload["internal_marker"] == EXPECTED_SUBAGENT_INTERNAL_MARKER + assert payload["parent_visible_marker"] == EXPECTED_PARENT_VISIBLE_CHILD_RESULT + + +def _assert_source_handoff_resource(sample_id: UUID) -> None: + resources = _require_named_resources( + sample_id, + prefix=HANDOFF_RESOURCE_NAME, + suffix="", + expected_count=1, + ) + payload = json.loads(read_resource_bytes(resources[0])) + assert payload["producer_slug"] == EXPECTED_RESOURCE_HANDOFF.producer_slug + assert payload["consumer_slug"] == EXPECTED_RESOURCE_HANDOFF.consumer_slug + assert payload["source_doc"] == SEEDED_SOURCE_DOC_NAME + + def _assert_minif2f_artifacts(sample_id: UUID) -> None: """Every MiniF2F leaf persists a Lean proof artifact with the smoke theorem.""" resources = _require_named_resources( @@ -330,22 +456,22 @@ def _after(child: str, parents: list[str]) -> None: f"{p}.completed_at ({p_exec.completed_at})" ) - _after("d_join", ["d_left", "d_right"]) - _after("d_left", ["d_root"]) - _after("d_right", ["d_root"]) - _after("l_2", ["l_1"]) - _after("l_3", ["l_2"]) + _after(ARTIFACT_SUMMARY_SLUG, [HANDOFF_VERIFY_SLUG, PRIMARY_ARTIFACT_SLUG]) + _after(HANDOFF_VERIFY_SLUG, [SOURCE_REVIEW_SLUG]) + _after(PRIMARY_ARTIFACT_SLUG, [SOURCE_REVIEW_SLUG]) + _after(ENV_PROBE_SLUG, [METADATA_REVIEW_SLUG]) + _after(METADATA_VALIDATE_SLUG, [ENV_PROBE_SLUG]) SMOKE_DIRECT_EDGES = ( - ("d_root", "d_left"), - ("d_root", "d_right"), - ("d_left", "d_join"), - ("d_right", "d_join"), - ("l_1", "l_2"), - ("l_2", "l_3"), + (SOURCE_REVIEW_SLUG, HANDOFF_VERIFY_SLUG), + (SOURCE_REVIEW_SLUG, PRIMARY_ARTIFACT_SLUG), + (HANDOFF_VERIFY_SLUG, ARTIFACT_SUMMARY_SLUG), + (PRIMARY_ARTIFACT_SLUG, ARTIFACT_SUMMARY_SLUG), + (METADATA_REVIEW_SLUG, ENV_PROBE_SLUG), + (ENV_PROBE_SLUG, METADATA_VALIDATE_SLUG), ) -SMOKE_NESTED_EDGES = (("l_2_a", "l_2_b"),) +SMOKE_NESTED_EDGES = ((NESTED_INSPECT_SLUG, NESTED_VERIFY_SLUG),) def _assert_sample_runtime_event_stream( @@ -374,32 +500,35 @@ def _assert_sample_runtime_event_stream( ) expected_terminal_status_by_slug = {slug: "completed" for slug in expected_task_slugs} if profile == "sad": - expected_terminal_status_by_slug["l_2"] = "failed" - expected_terminal_status_by_slug["l_3"] = "blocked" + expected_terminal_status_by_slug[ENV_PROBE_SLUG] = "failed" + expected_terminal_status_by_slug[METADATA_VALIDATE_SLUG] = "blocked" expected_workers = {slug: f"{worker_prefix}-smoke-leaf" for slug in expected_task_slugs} expected_workers[root_slug] = root_worker_slug if profile == "happy": - expected_workers["l_2"] = f"{worker_prefix}-smoke-recursive-worker" - expected_workers["l_2_a"] = f"{worker_prefix}-smoke-leaf" - expected_workers["l_2_b"] = f"{worker_prefix}-smoke-leaf" + expected_workers[ENV_PROBE_SLUG] = f"{worker_prefix}-smoke-recursive-worker" + expected_workers[NESTED_INSPECT_SLUG] = f"{worker_prefix}-smoke-leaf" + expected_workers[NESTED_VERIFY_SLUG] = f"{worker_prefix}-smoke-leaf" else: - expected_workers["l_2"] = f"{worker_prefix}-smoke-leaf-failing" + expected_workers[ENV_PROBE_SLUG] = f"{worker_prefix}-smoke-leaf-failing" assert snapshot.status_sequence == expected_status_sequence assert snapshot.task_added_slugs == expected_task_slugs assert snapshot.edge_added_pairs == expected_edge_pairs assert snapshot.worker_added_by_task_slug == expected_workers assert set(snapshot.sandbox_added_by_task_slug) == set(expected_task_slugs) - assert set(snapshot.evaluator_added_by_task_slug) == {root_slug} + assert set(snapshot.evaluator_added_by_task_slug) == {root_slug, HANDOFF_VERIFY_SLUG} assert set(snapshot.evaluator_added_by_task_slug[root_slug]) == {"default", "post-root"} + assert set(snapshot.evaluator_added_by_task_slug[HANDOFF_VERIFY_SLUG]) == { + "smoke-resource-handoff" + } assert snapshot.task_terminal_status_by_slug == expected_terminal_status_by_slug if profile == "happy": - assert "l_2_a" in snapshot.task_added_slugs - assert "l_2_b" in snapshot.task_added_slugs + assert NESTED_INSPECT_SLUG in snapshot.task_added_slugs + assert NESTED_VERIFY_SLUG in snapshot.task_added_slugs else: - assert "l_2_a" not in snapshot.task_added_slugs - assert "l_2_b" not in snapshot.task_added_slugs + assert NESTED_INSPECT_SLUG not in snapshot.task_added_slugs + assert NESTED_VERIFY_SLUG not in snapshot.task_added_slugs _assert_sample_runtime_event_order(snapshot, root_slug=root_slug) @@ -502,7 +631,7 @@ def _assert_experiment_membership(experiment: str, sample_ids: list[UUID]) -> No def _assert_sadpath_graph_cascade(sample_id: UUID) -> None: - """Canonical sad path: parent plans, l_2 fails, l_3 blocks, independent leaves complete.""" + """Canonical sad path: parent plans, environment-probe fails, metadata-validate blocks.""" snapshot = require_run_snapshot(sample_id) tasks = list(snapshot.tasks.values()) leaves = [task for task in tasks if task.level > 0] @@ -513,13 +642,17 @@ def _assert_sadpath_graph_cascade(sample_id: UUID) -> None: "parent task should complete after planning; child failure is represented " f"on the failing child and run terminal status, got {root_tasks[0].status}" ) - assert by_slug["l_2"].status == FAILED, f"l_2 expected FAILED, got {by_slug['l_2'].status}" - assert by_slug["l_3"].status == BLOCKED, f"l_3 expected BLOCKED, got {by_slug['l_3'].status}" - assert by_slug["l_3"].started_at is None, "blocked l_3 should never start" - assert not snapshot.executions_by_task.get(by_slug["l_3"].id), ( - "blocked l_3 should not have execution attempts" + failed = by_slug[ENV_PROBE_SLUG] + blocked = by_slug[METADATA_VALIDATE_SLUG] + assert failed.status == FAILED, f"{ENV_PROBE_SLUG} expected FAILED, got {failed.status}" + assert blocked.status == BLOCKED, ( + f"{METADATA_VALIDATE_SLUG} expected BLOCKED, got {blocked.status}" ) - for slug in set(EXPECTED_SUBTASK_SLUGS) - {"l_2", "l_3"}: + assert blocked.started_at is None, f"blocked {METADATA_VALIDATE_SLUG} should never start" + assert not snapshot.executions_by_task.get(blocked.id), ( + f"blocked {METADATA_VALIDATE_SLUG} should not have execution attempts" + ) + for slug in set(EXPECTED_SUBTASK_SLUGS) - {ENV_PROBE_SLUG, METADATA_VALIDATE_SLUG}: assert by_slug[slug].status == COMPLETED, ( f"{slug} expected COMPLETED, got {by_slug[slug].status}" ) @@ -536,7 +669,7 @@ def _assert_sadpath_partial_artifact(sample_id: UUID) -> None: break time.sleep(2) assert len(partials) == 1, ( - f"expected 1 partial artifact from l_2 (partial work must persist on " + f"expected 1 partial artifact from {ENV_PROBE_SLUG} (partial work must persist on " f"FAILED leaf), got {len(partials)}" ) r = partials[0] @@ -570,16 +703,20 @@ def _assert_sadpath_thread_messages(sample_id: UUID) -> None: assert thread is not None, "no smoke-completion thread created" msgs = sorted(thread.messages, key=lambda msg: msg.sequence_num) assert len(msgs) == 7, ( - f"expected 7 completion messages (l_2 failed, l_3 blocked), got {len(msgs)}" + f"expected 7 completion messages ({ENV_PROBE_SLUG} failed, " + f"{METADATA_VALIDATE_SLUG} blocked), got {len(msgs)}" ) from_slugs = {m.from_agent_id.removeprefix("leaf-") for m in msgs} - assert "l_2" not in from_slugs, ( - f"l_2 sent a completion message despite suppression: {from_slugs}" + assert ENV_PROBE_SLUG not in from_slugs, ( + f"{ENV_PROBE_SLUG} sent a completion message despite suppression: {from_slugs}" ) - assert "l_3" not in from_slugs, ( - f"l_3 sent a completion message despite being blocked: {from_slugs}" + assert METADATA_VALIDATE_SLUG not in from_slugs, ( + f"{METADATA_VALIDATE_SLUG} sent a completion message despite being blocked: {from_slugs}" ) - assert from_slugs == set(EXPECTED_SUBTASK_SLUGS) - {"l_2", "l_3"} + assert from_slugs == set(EXPECTED_SUBTASK_SLUGS) - { + ENV_PROBE_SLUG, + METADATA_VALIDATE_SLUG, + } def _assert_sadpath_evaluation(sample_id: UUID) -> None: diff --git a/tests/e2e/test_minif2f_smoke.py b/tests/e2e/test_minif2f_smoke.py index 0bf659953..09a26167a 100644 --- a/tests/e2e/test_minif2f_smoke.py +++ b/tests/e2e/test_minif2f_smoke.py @@ -14,8 +14,10 @@ from tests.e2e._asserts import ( _assert_blob_roundtrip, _assert_experiment_membership, + _assert_handoff_task_evaluation, _assert_minif2f_artifacts, _assert_run_evaluation, + _assert_rl_episode_view, _assert_sample_graph, _assert_sample_resources, _assert_run_turn_counts, @@ -27,8 +29,10 @@ _assert_sadpath_thread_messages, _assert_sandbox_command_wal, _assert_sandbox_lifecycle_events, + _assert_source_handoff_resource, _assert_temporal_ordering, _assert_thread_messages_ordered, + _assert_toolkit_probe_resources, wait_for_terminal_status, ) from tests.e2e._submit import submit_experiment_samples @@ -56,6 +60,14 @@ def _smoke_slots(group_size: int) -> list[SmokeSlot]: @pytest.mark.e2e +@pytest.mark.xfail( + reason=( + "Real-sandbox smoke currently exposes E2B/runtime bugs: private template " + "ergon-minif2f-v1 may be unavailable, public sandbox calls do not emit " + "command WAL, and E2B detach uses an SDK close method not present locally." + ), + strict=False, +) @pytest.mark.asyncio async def test_smoke_experiment_group(tmp_path: pathlib.Path) -> None: experiment = f"ci-smoke-{ENV}-{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%S')}" @@ -118,8 +130,15 @@ def _assert_happy_run(rid) -> None: _assert_run_turn_counts(rid) _assert_thread_messages_ordered(rid) _assert_blob_roundtrip(rid) + _assert_toolkit_probe_resources( + rid, + expected_toolkit_suffix="MiniF2FToolkit", + ) + _assert_source_handoff_resource(rid) _assert_minif2f_artifacts(rid) _assert_run_evaluation(rid) + _assert_handoff_task_evaluation(rid) + _assert_rl_episode_view(rid) _assert_sandbox_lifecycle_events(rid) _assert_sandbox_command_wal(rid) _assert_temporal_ordering(rid) diff --git a/tests/e2e/test_researchrubrics_smoke.py b/tests/e2e/test_researchrubrics_smoke.py index 32bfbaf74..6c8d7d53e 100644 --- a/tests/e2e/test_researchrubrics_smoke.py +++ b/tests/e2e/test_researchrubrics_smoke.py @@ -4,7 +4,7 @@ three runs: two happy-path runs plus one sad-path run. - ``happy`` uses the old fully-completing smoke worker. -- ``sad`` routes ``l_2`` to a failing leaf; ``l_3`` remains blocked. +- ``sad`` routes ``environment-probe`` to a failing leaf; ``metadata-validate`` remains blocked. Experiment group-level: ``_assert_experiment_membership`` checks all submitted runs are visible through the experiment-group harness endpoint. Playwright subprocess runs at the @@ -26,7 +26,9 @@ from tests.e2e._asserts import ( _assert_blob_roundtrip, _assert_experiment_membership, + _assert_handoff_task_evaluation, _assert_run_evaluation, + _assert_rl_episode_view, _assert_sample_graph, _assert_sample_resources, _assert_run_turn_counts, @@ -38,8 +40,10 @@ _assert_sadpath_thread_messages, _assert_sandbox_command_wal, _assert_sandbox_lifecycle_events, + _assert_source_handoff_resource, _assert_temporal_ordering, _assert_thread_messages_ordered, + _assert_toolkit_probe_resources, wait_for_terminal_status, ) from tests.e2e._submit import submit_experiment_samples @@ -68,6 +72,14 @@ def _smoke_slots(group_size: int) -> list[SmokeSlot]: @pytest.mark.e2e +@pytest.mark.xfail( + reason=( + "Real-sandbox smoke currently exposes E2B/runtime bugs: private template " + "ergon-research-v1 may be unavailable, public sandbox calls do not emit " + "command WAL, and E2B detach uses an SDK close method not present locally." + ), + strict=False, +) @pytest.mark.asyncio async def test_smoke_experiment_group(tmp_path: pathlib.Path) -> None: experiment = f"ci-smoke-{ENV}-{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%S')}" @@ -130,7 +142,14 @@ def _assert_happy_run(rid) -> None: _assert_run_turn_counts(rid) _assert_thread_messages_ordered(rid) _assert_blob_roundtrip(rid) + _assert_toolkit_probe_resources( + rid, + expected_toolkit_suffix="ResearchRubricsToolkit", + ) + _assert_source_handoff_resource(rid) _assert_run_evaluation(rid) + _assert_handoff_task_evaluation(rid) + _assert_rl_episode_view(rid) _assert_sandbox_lifecycle_events(rid) _assert_sandbox_command_wal(rid) _assert_temporal_ordering(rid) diff --git a/tests/e2e/test_swebench_smoke.py b/tests/e2e/test_swebench_smoke.py index f19298853..a8ac35407 100644 --- a/tests/e2e/test_swebench_smoke.py +++ b/tests/e2e/test_swebench_smoke.py @@ -14,7 +14,9 @@ from tests.e2e._asserts import ( _assert_blob_roundtrip, _assert_experiment_membership, + _assert_handoff_task_evaluation, _assert_run_evaluation, + _assert_rl_episode_view, _assert_sample_graph, _assert_sample_resources, _assert_run_turn_counts, @@ -26,9 +28,11 @@ _assert_sadpath_thread_messages, _assert_sandbox_command_wal, _assert_sandbox_lifecycle_events, + _assert_source_handoff_resource, _assert_swebench_artifacts, _assert_temporal_ordering, _assert_thread_messages_ordered, + _assert_toolkit_probe_resources, wait_for_terminal_status, ) from tests.e2e._submit import submit_experiment_samples @@ -61,6 +65,14 @@ def _smoke_slots(group_size: int) -> list[SmokeSlot]: @pytest.mark.e2e +@pytest.mark.xfail( + reason=( + "Real-sandbox smoke currently exposes E2B/runtime bugs: private template " + "ergon-swebench-v1 may be unavailable, public sandbox calls do not emit " + "command WAL, and E2B detach uses an SDK close method not present locally." + ), + strict=False, +) @pytest.mark.asyncio async def test_smoke_experiment_group(tmp_path: pathlib.Path) -> None: experiment = f"ci-smoke-{ENV}-{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%S')}" @@ -123,8 +135,15 @@ def _assert_happy_run(rid) -> None: _assert_run_turn_counts(rid) _assert_thread_messages_ordered(rid) _assert_blob_roundtrip(rid) + _assert_toolkit_probe_resources( + rid, + expected_toolkit_suffix="SWEBenchToolkit", + ) + _assert_source_handoff_resource(rid) _assert_swebench_artifacts(rid) _assert_run_evaluation(rid) + _assert_handoff_task_evaluation(rid) + _assert_rl_episode_view(rid) _assert_sandbox_lifecycle_events(rid) _assert_sandbox_command_wal(rid) _assert_temporal_ordering(rid) diff --git a/tests/fixtures/smoke_components/benchmarks.py b/tests/fixtures/smoke_components/benchmarks.py index 2fd17fb8b..a5dae7885 100644 --- a/tests/fixtures/smoke_components/benchmarks.py +++ b/tests/fixtures/smoke_components/benchmarks.py @@ -10,12 +10,15 @@ from collections.abc import Iterator, Sequence from typing import ClassVar, Literal +from ergon_builtins.benchmarks.gdpeval.sandbox import GDPEvalSandbox +from ergon_builtins.benchmarks.minif2f.sandbox import LeanSandbox +from ergon_builtins.benchmarks.researchrubrics.sandbox import ResearchE2BSandbox +from ergon_builtins.benchmarks.swebench_verified.sandbox import SWEBenchSandbox from ergon_core.api import Environment, Sample from ergon_core.api.task import Task from ergon_core.api.worker import Worker from ergon_core.core.shared.json_types import JsonObject from pydantic import BaseModel -from tests.fixtures.smoke_components.sandbox import SmokePublicSandbox from tests.fixtures.smoke_components.workers.minif2f_smoke import ( MiniF2FFailingLeafWorker, MiniF2FRecursiveSmokeWorker, @@ -170,20 +173,21 @@ class ResearchRubricsSmokeEnvironment(_SingleTaskSmokeEnvironment): Returns a concrete ``ResearchRubricsSmokeTask`` with inline ``evaluators``, so the smoke fixture exercises the v2 object-bound path that the production - ResearchRubrics environment now uses. ``sandbox`` uses the test-owned - public wrapper over ``SmokeSandboxManager`` so eval-side - ``Task.from_definition(..., sandbox_id=...)`` can attach a live - runtime without reaching E2B. + ResearchRubrics environment now uses. ``sandbox`` uses the real + benchmark sandbox class so the smoke fixture mirrors production + posture while the test harness controls the runtime manager. """ environment_slug: ClassVar[str] = "researchrubrics" default_worker_slug: ClassVar[str] = "researchrubrics-smoke-worker" task_slug: ClassVar[str] = "smoke-001" - task_description: ClassVar[str] = "Write a short smoke-test research report." + task_description: ClassVar[str] = ( + "Review the supplied research notes and produce a concise evidence-backed report." + ) task_payload: ClassVar[JsonObject] = { "sample_id": "smoke-001", "domain": "smoke", - "prompt": "Write a short smoke-test research report.", + "prompt": "Review the supplied research notes and produce a concise evidence-backed report.", "rubrics": [ { "criterion": "Report contains the expected smoke-test marker.", @@ -216,7 +220,7 @@ def _tasks(self) -> Sequence[Task[ResearchRubricsTaskPayload]]: description=self.task_description, task_payload=payload, worker=self._make_worker(), - sandbox=SmokePublicSandbox(), + sandbox=ResearchE2BSandbox(), evaluators=( ResearchRubricsSmokeRubric(name="default"), SmokePostRootTimingRubric(name="post-root"), @@ -240,16 +244,15 @@ class MiniF2FSmokeEnvironment(_SingleTaskSmokeEnvironment): Returns a concrete ``MiniF2FSmokeTask`` with inline ``evaluators``, so the smoke fixture exercises the v2 object-bound path that the production MiniF2F environment now uses. - ``sandbox`` uses the test-owned public wrapper over - ``SmokeSandboxManager`` so eval-side - ``Task.from_definition(..., sandbox_id=...)`` can attach a live - runtime without reaching E2B. + ``sandbox`` uses the real benchmark sandbox class so the smoke fixture + mirrors production posture while the test harness controls the runtime + manager. """ environment_slug: ClassVar[str] = "minif2f" default_worker_slug: ClassVar[str] = "minif2f-smoke-worker" task_slug: ClassVar[str] = "mathd_algebra_478" - task_description: ClassVar[str] = "Prove the smoke_trivial theorem in Lean." + task_description: ClassVar[str] = "Verify the supplied Lean theorem and record proof evidence." task_payload: ClassVar[JsonObject] = { "name": "mathd_algebra_478", "informal_statement": "Smoke theorem used by the canonical E2E fixture.", @@ -277,7 +280,7 @@ def _tasks(self) -> Sequence[Task[MiniF2FTaskPayload]]: description=self.task_description, task_payload=payload, worker=self._make_worker(), - sandbox=SmokePublicSandbox(), + sandbox=LeanSandbox(), evaluators=( MiniF2FSmokeRubric(name="default"), SmokePostRootTimingRubric(name="post-root"), @@ -301,16 +304,17 @@ class SweBenchSmokeEnvironment(_SingleTaskSmokeEnvironment): Returns a concrete ``SweBenchSmokeTask`` with inline ``evaluators``, so the smoke fixture exercises the v2 object-bound path that the production SWE-Bench environment now uses. - ``sandbox`` uses the test-owned public wrapper over - ``SmokeSandboxManager`` so eval-side - ``Task.from_definition(..., sandbox_id=...)`` can attach a live - runtime without reaching E2B. + ``sandbox`` uses the real benchmark sandbox class so the smoke fixture + mirrors production posture while the test harness controls the runtime + manager. """ environment_slug: ClassVar[str] = "swebench-verified" default_worker_slug: ClassVar[str] = "swebench-smoke-worker" task_slug: ClassVar[str] = "astropy__astropy-12907" - task_description: ClassVar[str] = "Create the simple Python add() patch used by smoke tests." + task_description: ClassVar[str] = ( + "Inspect the Python issue and produce a minimal source patch with evidence." + ) task_payload: ClassVar[JsonObject] = { "instance_id": "astropy__astropy-12907", "repo": "smoke/repo", @@ -344,7 +348,7 @@ def _tasks(self) -> Sequence[Task[SWEBenchTaskPayload]]: description=self.task_description, task_payload=payload, worker=self._make_worker(), - sandbox=SmokePublicSandbox(), + sandbox=SWEBenchSandbox(), evaluators=( SweBenchSmokeRubric(name="default"), SmokePostRootTimingRubric(name="post-root"), @@ -377,7 +381,9 @@ class GDPEvalSmokeEnvironment(_SingleTaskSmokeEnvironment): environment_slug: ClassVar[str] = "gdpeval" default_worker_slug: ClassVar[str] = "gdpeval-smoke-worker" task_slug: ClassVar[str] = "gdpeval-smoke-001" - task_description: ClassVar[str] = "Process the reference documents and write outputs." + task_description: ClassVar[str] = ( + "Process the reference documents and write a structured output bundle." + ) task_payload: ClassVar[JsonObject] = { "task_id": "gdpeval-smoke-001", "workflow_type": "document_processing", @@ -401,7 +407,7 @@ def _tasks(self) -> Sequence[Task[GDPEvalTaskPayload]]: description=self.task_description, task_payload=payload, worker=self._make_worker(), - sandbox=SmokePublicSandbox(), + sandbox=GDPEvalSandbox(), evaluators=(SmokePostRootTimingRubric(name="post-root"),), ) return [task] diff --git a/tests/fixtures/smoke_components/smoke_base/constants.py b/tests/fixtures/smoke_components/smoke_base/constants.py index 7680743a0..6ebb72077 100644 --- a/tests/fixtures/smoke_components/smoke_base/constants.py +++ b/tests/fixtures/smoke_components/smoke_base/constants.py @@ -1,45 +1,125 @@ """Immutable topology + slug constants for the canonical smoke DAG. -One place. ``SmokeWorkerBase``, ``SmokeCriterionBase``, every pytest -driver, and every Playwright spec import from here. Changing this tuple -is the only way to change smoke topology. +One place. ``SmokeWorkerBase``, ``SmokeCriterionBase``, every pytest driver, +and every Playwright spec import from here. Changing this tuple is the only +way to change smoke topology. -Shape: 4-node diamond + 3-node line + 2 singletons = 9 leaves. +Shape: 4-node diamond + 3-node line + 2 singletons = 9 direct children. - d_root ── d_left ─┐ - └─ d_right ─┴─> d_join + source-review -> handoff-verify -. + `> primary-artifact -> artifact-summary - l_1 ──> l_2 ──> l_3 + metadata-review -> environment-probe -> metadata-validate - s_a s_b + evidence-artifact completion-marker """ from collections.abc import Sequence +from dataclasses import dataclass + +SOURCE_REVIEW_SLUG = "source-review" +HANDOFF_VERIFY_SLUG = "handoff-verify" +PRIMARY_ARTIFACT_SLUG = "primary-artifact" +ARTIFACT_SUMMARY_SLUG = "artifact-summary" +METADATA_REVIEW_SLUG = "metadata-review" +ENV_PROBE_SLUG = "environment-probe" +METADATA_VALIDATE_SLUG = "metadata-validate" +EVIDENCE_ARTIFACT_SLUG = "evidence-artifact" +COMPLETION_MARKER_SLUG = "completion-marker" + +NESTED_INSPECT_SLUG = "nested-input-review" +NESTED_VERIFY_SLUG = "nested-verification" +NESTED_LINE_SLUGS: tuple[str, ...] = (NESTED_INSPECT_SLUG, NESTED_VERIFY_SLUG) EXPECTED_SUBTASK_SLUGS: tuple[str, ...] = ( - "d_root", - "d_left", - "d_right", - "d_join", - "l_1", - "l_2", - "l_3", - "s_a", - "s_b", + SOURCE_REVIEW_SLUG, + HANDOFF_VERIFY_SLUG, + PRIMARY_ARTIFACT_SLUG, + ARTIFACT_SUMMARY_SLUG, + METADATA_REVIEW_SLUG, + ENV_PROBE_SLUG, + METADATA_VALIDATE_SLUG, + EVIDENCE_ARTIFACT_SLUG, + COMPLETION_MARKER_SLUG, ) -# (slug, depends_on_slugs, description) — shape of the DAG in one place. +# (slug, depends_on_slugs, description) - shape of the DAG in one place. # Order is authoritative: ``SmokeWorkerBase.execute`` iterates this tuple # in-order when spawning object-bound child tasks. Leaves appear before anything # that depends on them so slug-level forward refs are avoided. SUBTASK_GRAPH: Sequence[tuple[str, tuple[str, ...], str]] = ( - ("d_root", (), "Diamond root"), - ("d_left", ("d_root",), "Diamond left arm"), - ("d_right", ("d_root",), "Diamond right arm"), - ("d_join", ("d_left", "d_right"), "Diamond join (two-parent fan-in)"), - ("l_1", (), "Line node 1"), - ("l_2", ("l_1",), "Line node 2"), - ("l_3", ("l_2",), "Line node 3"), - ("s_a", (), "Singleton A"), - ("s_b", (), "Singleton B"), + ( + SOURCE_REVIEW_SLUG, + (), + "Review the seeded source brief and publish the handoff JSON.", + ), + ( + HANDOFF_VERIFY_SLUG, + (SOURCE_REVIEW_SLUG,), + "Read the source-review handoff, confirm the coordination thread, and validate the resource.", + ), + ( + PRIMARY_ARTIFACT_SLUG, + (SOURCE_REVIEW_SLUG,), + "Produce the primary benchmark artifact from the reviewed source brief.", + ), + ( + ARTIFACT_SUMMARY_SLUG, + (HANDOFF_VERIFY_SLUG, PRIMARY_ARTIFACT_SLUG), + "Summarise the verified handoff and primary artifact after both parents finish.", + ), + ( + METADATA_REVIEW_SLUG, + (), + "Inspect sample metadata and record the expected smoke annotations.", + ), + ( + ENV_PROBE_SLUG, + (METADATA_REVIEW_SLUG,), + "Run the recursive environment probe and plan nested sandbox checks.", + ), + ( + METADATA_VALIDATE_SLUG, + (ENV_PROBE_SLUG,), + "Validate metadata after the recursive environment probe completes.", + ), + ( + EVIDENCE_ARTIFACT_SLUG, + (), + "Write an independent evidence artifact for resource persistence.", + ), + ( + COMPLETION_MARKER_SLUG, + (), + "Emit a final independent completion marker artifact.", + ), +) + +SEEDED_SOURCE_DOC_NAME = "smoke_source_brief.md" +SMOKE_THREAD_TOPIC = "smoke-coordination" +SMOKE_COMPLETION_THREAD_TOPIC = "smoke-completion" +HANDOFF_RESOURCE_NAME = "handoff_source_review.json" +SMOKE_OUTPUT_DIR = "/workspace/final_output" +SMOKE_WORKSPACE_DIR = "/workspace" +SMOKE_REPO_DIR = "/workspace/repo" +SMOKE_HANDOFF_PRODUCER_SLUG = SOURCE_REVIEW_SLUG +SMOKE_HANDOFF_CONSUMER_SLUG = HANDOFF_VERIFY_SLUG + +EXPECTED_SMOKE_TOKEN_IDS: list[int] = [101, 202, 303] +EXPECTED_SMOKE_LOGPROBS: list[float] = [-0.10, -0.20, -0.30] +EXPECTED_SUBAGENT_INTERNAL_MARKER = "SMOKE_CHILD_INTERNAL_THINKING" +EXPECTED_PARENT_VISIBLE_CHILD_RESULT = "SMOKE_CHILD_RESULT_PARENT_VISIBLE" + + +@dataclass(frozen=True) +class ExpectedResourceHandoff: + producer_slug: str + consumer_slug: str + resource_name: str + + +EXPECTED_RESOURCE_HANDOFF = ExpectedResourceHandoff( + producer_slug=SOURCE_REVIEW_SLUG, + consumer_slug=HANDOFF_VERIFY_SLUG, + resource_name=HANDOFF_RESOURCE_NAME, ) diff --git a/tests/fixtures/smoke_components/smoke_base/criterion_base.py b/tests/fixtures/smoke_components/smoke_base/criterion_base.py index 8dafec35c..2f992665d 100644 --- a/tests/fixtures/smoke_components/smoke_base/criterion_base.py +++ b/tests/fixtures/smoke_components/smoke_base/criterion_base.py @@ -174,8 +174,8 @@ async def _artifact_children( ) -> list[SampleGraphNode]: """Return leaf descendants that should publish probe/artifact resources. - The happy smoke path routes direct child ``l_2`` to a recursive worker. - ``l_2`` is still part of the direct-child topology check, but its + The happy smoke path routes direct child ``environment-probe`` to a recursive worker. + ``environment-probe`` is still part of the direct-child topology check, but its nested children are the artifact-producing leaves. """ with get_session() as session: diff --git a/tests/fixtures/smoke_components/smoke_base/dynamic_tasks.py b/tests/fixtures/smoke_components/smoke_base/dynamic_tasks.py index 32ae8042c..51a40ce4a 100644 --- a/tests/fixtures/smoke_components/smoke_base/dynamic_tasks.py +++ b/tests/fixtures/smoke_components/smoke_base/dynamic_tasks.py @@ -3,9 +3,9 @@ from pydantic import BaseModel, Field from ergon_core.api import Task +from ergon_core.api.rubric import Evaluator from ergon_core.api.worker import Worker from ergon_core.core.persistence.shared.types import AssignedWorkerSlug, TaskSlug -from tests.fixtures.smoke_components.sandbox import SmokePublicSandbox class SmokeChildTaskSpec(BaseModel): @@ -15,8 +15,9 @@ class SmokeChildTaskSpec(BaseModel): description: str = Field(min_length=1) assigned_worker_slug: AssignedWorkerSlug depends_on: list[TaskSlug] = Field(default_factory=list) + evaluators: tuple["Evaluator", ...] = () - model_config = {"frozen": True} + model_config = {"frozen": True, "arbitrary_types_allowed": True} def smoke_worker_for_slug(worker_slug: str, *, model: str | None) -> Worker: @@ -26,6 +27,23 @@ def smoke_worker_for_slug(worker_slug: str, *, model: str | None) -> Worker: return worker_cls(name=worker_slug, model=model) +def _copy_sandbox_config(parent_task: Task): + sandbox_payload = parent_task.sandbox.model_dump(mode="python") + sandbox_payload.pop("_type", None) + return type(parent_task.sandbox).model_validate(sandbox_payload) + + +def smoke_evaluators_for_slug(slug: str) -> tuple["Evaluator", ...]: + from tests.fixtures.smoke_components.smoke_base.constants import HANDOFF_VERIFY_SLUG + from tests.fixtures.smoke_components.smoke_base.handoff_evaluator import ( + SmokeResourceHandoffRubric, + ) + + if slug == HANDOFF_VERIFY_SLUG: + return (SmokeResourceHandoffRubric(name="smoke-resource-handoff"),) + return () + + def smoke_task_from_spec( *, parent_task: Task, @@ -33,6 +51,8 @@ def smoke_task_from_spec( model: str | None, ) -> Task: worker_slug = str(spec.assigned_worker_slug) + if parent_task.sandbox is None: + raise ValueError("parent task must define a sandbox before spawning smoke child tasks") return Task( task_slug=str(spec.task_slug), instance_key=parent_task.instance_key, @@ -40,5 +60,6 @@ def smoke_task_from_spec( parent_task_slug=parent_task.task_slug, dependency_task_slugs=tuple(str(dep) for dep in spec.depends_on), worker=smoke_worker_for_slug(worker_slug, model=model), - sandbox=SmokePublicSandbox(), + sandbox=_copy_sandbox_config(parent_task), + evaluators=spec.evaluators, ) diff --git a/tests/fixtures/smoke_components/smoke_base/handoff_evaluator.py b/tests/fixtures/smoke_components/smoke_base/handoff_evaluator.py new file mode 100644 index 000000000..8add3fc26 --- /dev/null +++ b/tests/fixtures/smoke_components/smoke_base/handoff_evaluator.py @@ -0,0 +1,49 @@ +"""Dynamic smoke evaluator for the resource handoff consumer task.""" + +from typing import ClassVar + +from pydantic import Field, model_validator + +from ergon_core.api.criterion import Criterion, CriterionContext, CriterionOutcome +from ergon_core.api.rubric import Rubric +from tests.fixtures.smoke_components.smoke_base.constants import ( + EXPECTED_RESOURCE_HANDOFF, + HANDOFF_RESOURCE_NAME, +) + + +class SmokeResourceHandoffCriterion(Criterion): + """Assert the consumer leaf observed the producer handoff resource.""" + + type_slug: ClassVar[str] = "smoke-resource-handoff-criterion" + + async def evaluate(self, context: CriterionContext) -> CriterionOutcome: + handoff_probe = context.worker_result.metadata.get("handoff_probe", {}) + passed = bool( + handoff_probe.get("checked") + and handoff_probe.get("ok") + and handoff_probe.get("resource_name") == HANDOFF_RESOURCE_NAME + and handoff_probe.get("producer_slug") == EXPECTED_RESOURCE_HANDOFF.producer_slug + ) + return CriterionOutcome( + slug=self.slug, + name=self.slug, + score=1.0 if passed else 0.0, + passed=passed, + weight=self.weight, + feedback="handoff resource was visible to consumer" if passed else repr(handoff_probe), + ) + + +class SmokeResourceHandoffRubric(Rubric): + """Evaluator attached only to the dynamic handoff consumer task.""" + + type_slug: ClassVar[str] = "smoke-resource-handoff-evaluator" + name: str = "smoke-resource-handoff-evaluator" + criteria: tuple[Criterion, ...] = Field(default_factory=tuple, exclude=True) + + @model_validator(mode="after") + def _build_criterion(self) -> "SmokeResourceHandoffRubric": + if not self.criteria: + self.criteria = (SmokeResourceHandoffCriterion(slug="smoke-resource-handoff"),) + return self diff --git a/tests/fixtures/smoke_components/smoke_base/leaf_base.py b/tests/fixtures/smoke_components/smoke_base/leaf_base.py index 35d182e4b..863edb2e9 100644 --- a/tests/fixtures/smoke_components/smoke_base/leaf_base.py +++ b/tests/fixtures/smoke_components/smoke_base/leaf_base.py @@ -20,21 +20,33 @@ See docs/superpowers/plans/test-refactor/01-fixtures.md §2.4. """ +import json from collections.abc import AsyncGenerator, Mapping from typing import Any, ClassVar from uuid import UUID from ergon_core.api import Task, Worker, WorkerContext, WorkerStreamItem +from ergon_core.api.sandbox.runtime import CommandResult from ergon_core.api.worker import WorkerOutput from ergon_core.core.persistence.graph.models import SampleGraphNode from ergon_core.core.persistence.shared.db import get_session -from ergon_core.core.infrastructure.sandbox.instrumentation import InstrumentedSandbox from ergon_core.core.application.communication.models import CreateMessageRequest from ergon_core.core.application.communication.service import ( communication_service, ) -from ergon_core.core.shared.settings import settings -from tests.fixtures.smoke_components.sandbox import SmokeSandboxManager +from tests.fixtures.smoke_components.smoke_base.constants import ( + EXPECTED_PARENT_VISIBLE_CHILD_RESULT, + EXPECTED_RESOURCE_HANDOFF, + EXPECTED_SMOKE_LOGPROBS, + EXPECTED_SMOKE_TOKEN_IDS, + EXPECTED_SUBAGENT_INTERNAL_MARKER, + HANDOFF_RESOURCE_NAME, + SEEDED_SOURCE_DOC_NAME, + SMOKE_OUTPUT_DIR, + SMOKE_REPO_DIR, + SMOKE_THREAD_TOPIC, + SMOKE_WORKSPACE_DIR, +) from tests.fixtures.smoke_components.smoke_base.metrics import smoke_assistant_chunk from tests.fixtures.smoke_components.smoke_base.subworker import ( SmokeSubworker, @@ -49,6 +61,7 @@ class BaseSmokeLeafWorker(Worker): # Subclasses bind a concrete SmokeSubworker implementation here. The # leaf's ``execute`` instantiates ``subworker_cls()`` and delegates. subworker_cls: ClassVar[type[SmokeSubworker]] + toolkit_type: ClassVar[str | None] = None # Driver asserts per-leaf context chunk count against this constant. # Sad-path leaves that raise inside subworker.work() emit fewer turns @@ -87,15 +100,31 @@ async def execute( ), ) - raw_sandbox = await SmokeSandboxManager().reconnect(context.sandbox_id) - sandbox = InstrumentedSandbox( - raw_sandbox, - SmokeSandboxManager._event_sink, - context.sample_id, - context.task_id or context.execution_id, - settings.otel_stdout_stderr_max_length, + if not task.sandbox.is_live: + raise RuntimeError(f"{type(task.sandbox).__name__} is not live for smoke leaf") + + sandbox = _PublicSandboxAdapter(task.sandbox) + await task.sandbox.run_command( + f"mkdir -p {SMOKE_OUTPUT_DIR} {SMOKE_WORKSPACE_DIR} {SMOKE_REPO_DIR}", + timeout=10, ) + await self._prepare_semantic_inputs(task, context) result = await self.subworker_cls().work(task_id=task_hex, sandbox=sandbox) + toolkit_probe = await self._run_builtin_toolkit_probe(task=task, context=context) + toolkit_probe.update( + { + "synthetic_token_ids": EXPECTED_SMOKE_TOKEN_IDS, + "synthetic_logprobs": EXPECTED_SMOKE_LOGPROBS, + "internal_marker": EXPECTED_SUBAGENT_INTERNAL_MARKER, + "parent_visible_marker": EXPECTED_PARENT_VISIBLE_CHILD_RESULT, + }, + ) + await task.sandbox.write_file( + f"{SMOKE_OUTPUT_DIR}/toolkit_probe_{task_hex}.json", + json.dumps(toolkit_probe, sort_keys=True), + ) + await self._maybe_publish_source_handoff(task=task, context=context) + handoff_probe = await self._maybe_verify_source_handoff(context=context) self._last_result = result # Post a one-line completion message to the shared @@ -118,9 +147,147 @@ async def execute( metadata={ "probe_exit_code": result.probe_exit_code, "file_path": result.file_path, + "toolkit_probe": toolkit_probe, + "handoff_probe": handoff_probe, }, ) + async def _prepare_semantic_inputs(self, task: Task, context: WorkerContext) -> None: + task_slug = self._lookup_task_slug(context.task_id) + if task_slug != EXPECTED_RESOURCE_HANDOFF.producer_slug: + return + await task.sandbox.write_file( + f"{SMOKE_OUTPUT_DIR}/{SEEDED_SOURCE_DOC_NAME}", + ( + "# Smoke Source Brief\n\n" + "Use this seeded brief to prove resource handoff and task-local " + "sandbox IO both work in canonical smoke.\n" + ), + ) + + async def _maybe_publish_source_handoff( + self, + *, + task: Task, + context: WorkerContext, + ) -> None: + task_slug = self._lookup_task_slug(context.task_id) + if task_slug != EXPECTED_RESOURCE_HANDOFF.producer_slug: + return + payload = { + "producer_slug": EXPECTED_RESOURCE_HANDOFF.producer_slug, + "consumer_slug": EXPECTED_RESOURCE_HANDOFF.consumer_slug, + "source_doc": SEEDED_SOURCE_DOC_NAME, + "parent_visible_result": EXPECTED_PARENT_VISIBLE_CHILD_RESULT, + } + await task.sandbox.write_file( + f"{SMOKE_OUTPUT_DIR}/{HANDOFF_RESOURCE_NAME}", + json.dumps(payload, sort_keys=True), + ) + await communication_service.save_message( + CreateMessageRequest( + sample_id=context.sample_id, + task_attempt_id=context.execution_id, + from_agent_id=f"leaf-{task_slug}", + to_agent_id=f"leaf-{EXPECTED_RESOURCE_HANDOFF.consumer_slug}", + thread_topic=SMOKE_THREAD_TOPIC, + content=f"handoff ready: {HANDOFF_RESOURCE_NAME}", + ), + ) + + async def _maybe_verify_source_handoff(self, *, context: WorkerContext) -> dict[str, Any]: + task_slug = self._lookup_task_slug(context.task_id) + if task_slug != EXPECTED_RESOURCE_HANDOFF.consumer_slug: + return {"checked": False} + + deadline_resources = await context.resources(name=HANDOFF_RESOURCE_NAME) + if not deadline_resources: + return {"checked": True, "ok": False, "reason": "handoff resource missing"} + payload = json.loads((await context.read_resource(deadline_resources[0].id)).decode()) + return { + "checked": True, + "ok": payload.get("consumer_slug") == EXPECTED_RESOURCE_HANDOFF.consumer_slug, + "resource_name": deadline_resources[0].name, + "producer_slug": payload.get("producer_slug"), + } + + async def _run_builtin_toolkit_probe( + self, + *, + task: Task, + context: WorkerContext, + ) -> dict[str, Any]: + del context + if self.toolkit_type is None: + return {"toolkit": None, "ok": True} + + if self.toolkit_type.endswith(":ResearchRubricsToolkit"): + from ergon_builtins.benchmarks.researchrubrics.toolkit import ( + ResearchRubricsToolkit, + ) + + toolkit = ResearchRubricsToolkit( + enable_web_browse=False, + workspace_root=SMOKE_WORKSPACE_DIR, + ) + tools = {tool.name: tool for tool in toolkit.tools(task.sandbox, task)} + write = await tools["write_report"].function( + "tmp/toolkit_probe.md", + "research toolkit smoke probe\n", + ) + read = await tools["read_report"].function("tmp/toolkit_probe.md") + return { + "toolkit": self.toolkit_type, + "ok": bool(write.ok and read.ok), + "marker": "research-toolkit", + } + + if self.toolkit_type.endswith(":MiniF2FToolkit"): + from ergon_builtins.benchmarks.minif2f.toolkit import MiniF2FToolkit + + toolkit = MiniF2FToolkit() + tools = {tool.name: tool for tool in toolkit.tools(task.sandbox, task)} + path = f"{SMOKE_WORKSPACE_DIR}/toolkit_probe.lean" + write = await tools["write_lean_file"].function( + path, + "theorem smoke_toolkit_probe : True := by trivial\n", + ) + verify = await tools["verify_lean_proof"].function(path) + return { + "toolkit": self.toolkit_type, + "ok": bool(write.success and verify.success and verify.verified), + "marker": "lean-toolkit", + } + + if self.toolkit_type.endswith(":SWEBenchToolkit"): + from ergon_builtins.benchmarks.swebench_verified.toolkit import ( + SWEBenchToolkit, + ) + + await task.sandbox.run_command(f"mkdir -p {SMOKE_REPO_DIR}", timeout=10) + toolkit = SWEBenchToolkit(repo_root=SMOKE_REPO_DIR) + tools = {tool.name: tool for tool in toolkit.tools(task.sandbox, task)} + edit = await tools["str_replace_editor"].function( + command="create", + path="toolkit_probe.py", + file_text="def smoke_toolkit_probe():\n return True\n", + ) + bash = await tools["bash"].function( + "python -m py_compile toolkit_probe.py", + timeout_sec=60, + ) + return { + "toolkit": self.toolkit_type, + "ok": bool(edit.ok and bash.exit_code == 0), + "marker": "swebench-toolkit", + } + + return { + "toolkit": self.toolkit_type, + "ok": True, + "marker": "unprobed-toolkit", + } + async def _send_completion_message( self, context: WorkerContext, @@ -136,7 +303,7 @@ async def _send_completion_message( ``SampleGraphNode.task_slug`` by ``context.task_id`` - ``to_agent_id``: ``"parent"`` - 9 messages per happy run, sequence_num 1..9 per-thread-monotonic - - 8 messages per sad run (l_2 suppresses this call; l_3 still runs) + - 8 messages per sad run (environment-probe suppresses this call; metadata-validate is blocked) """ task_slug = self._lookup_task_slug(context.task_id) await communication_service.save_message( @@ -169,3 +336,28 @@ def _lookup_task_slug(task_id: UUID | None) -> str: select(SampleGraphNode).where(SampleGraphNode.task_id == task_id) ).first() return node.task_slug if node is not None else f"node-{task_id.hex[:8]}" + + +class _PublicSandboxFiles: + def __init__(self, sandbox) -> None: + self._sandbox = sandbox + + async def write(self, path: str, content: bytes | str) -> None: + await self._sandbox.write_file(path, content) + + async def read(self, path: str) -> bytes: + return await self._sandbox.read_file(path) + + +class _PublicSandboxCommands: + def __init__(self, sandbox) -> None: + self._sandbox = sandbox + + async def run(self, command: str, *, timeout: int | None = None) -> CommandResult: + return await self._sandbox.run_command(command, timeout=timeout) + + +class _PublicSandboxAdapter: + def __init__(self, sandbox) -> None: + self.files = _PublicSandboxFiles(sandbox) + self.commands = _PublicSandboxCommands(sandbox) diff --git a/tests/fixtures/smoke_components/smoke_base/metrics.py b/tests/fixtures/smoke_components/smoke_base/metrics.py index fa6a293a1..b3adc80e3 100644 --- a/tests/fixtures/smoke_components/smoke_base/metrics.py +++ b/tests/fixtures/smoke_components/smoke_base/metrics.py @@ -4,6 +4,11 @@ AssistantTextPart, ContextPartChunk, ProviderTokenUsage, + TokenLogprob, +) +from tests.fixtures.smoke_components.smoke_base.constants import ( + EXPECTED_SMOKE_LOGPROBS, + EXPECTED_SMOKE_TOKEN_IDS, ) _SMOKE_TOKEN_COST_USD = 0.000001 @@ -14,6 +19,15 @@ def smoke_assistant_chunk(content: str) -> ContextPartChunk: token_count = max(1, len(content.split())) return ContextPartChunk( part=AssistantTextPart(content=content), + token_ids=EXPECTED_SMOKE_TOKEN_IDS, + logprobs=[ + TokenLogprob(token=f"smoke-{token_id}", logprob=logprob) + for token_id, logprob in zip( + EXPECTED_SMOKE_TOKEN_IDS, + EXPECTED_SMOKE_LOGPROBS, + strict=True, + ) + ], provider_usage=ProviderTokenUsage( completion_tokens=token_count, total_tokens=token_count, diff --git a/tests/fixtures/smoke_components/smoke_base/recursive.py b/tests/fixtures/smoke_components/smoke_base/recursive.py index 1d96eac91..1885e4241 100644 --- a/tests/fixtures/smoke_components/smoke_base/recursive.py +++ b/tests/fixtures/smoke_components/smoke_base/recursive.py @@ -1,9 +1,9 @@ """Recursive smoke worker used by happy-path E2E runs. -This worker is assigned to the top-level ``l_2`` node. It plans two +This worker is assigned to the top-level ``environment-probe`` node. It plans two nested leaf subtasks under itself, waits for them to complete, then sends the same completion-thread message shape as a normal leaf. The top-level -``l_3`` dependency therefore waits on a non-leaf dynamic task. +``metadata-validate`` dependency therefore waits on a non-leaf dynamic task. """ from collections.abc import AsyncGenerator @@ -17,22 +17,32 @@ from ergon_core.core.persistence.shared.types import AssignedWorkerSlug, TaskSlug from ergon_core.core.application.communication.models import CreateMessageRequest from ergon_core.core.application.communication.service import communication_service +from tests.fixtures.smoke_components.smoke_base.constants import ( + ENV_PROBE_SLUG, + NESTED_INSPECT_SLUG, + NESTED_LINE_SLUGS, + NESTED_VERIFY_SLUG, +) from tests.fixtures.smoke_components.smoke_base.dynamic_tasks import ( SmokeChildTaskSpec, + smoke_evaluators_for_slug, smoke_task_from_spec, ) from tests.fixtures.smoke_components.smoke_base.metrics import smoke_assistant_chunk from sqlmodel import select -NESTED_LINE_SLUGS: tuple[str, ...] = ("l_2_a", "l_2_b") NESTED_SUBTASK_GRAPH: tuple[tuple[str, tuple[str, ...], str], ...] = ( - ("l_2_a", (), "Nested line node 2a"), - ("l_2_b", ("l_2_a",), "Nested line node 2b"), + (NESTED_INSPECT_SLUG, (), "Nested input review for environment probe"), + ( + NESTED_VERIFY_SLUG, + (NESTED_INSPECT_SLUG,), + "Nested verification using nested input review", + ), ) class RecursiveSmokeWorkerBase(Worker): - """Plan and wait for a two-node nested line under ``l_2``.""" + """Plan and wait for a two-node nested line under ``environment-probe``.""" leaf_slug: ClassVar[str] RECURSIVE_TURN_COUNT: ClassVar[int] = 3 @@ -125,9 +135,9 @@ def _lookup_task_slug(task_id: UUID | None) -> str: class RecursiveSmokeWorkerMixin: - """Route top-level ``l_2`` to an env-specific recursive worker.""" + """Route top-level ``environment-probe`` to an env-specific recursive worker.""" - RECURSIVE_SLUGS: ClassVar[frozenset[str]] = frozenset({"l_2"}) + RECURSIVE_SLUGS: ClassVar[frozenset[str]] = frozenset({ENV_PROBE_SLUG}) RECURSIVE_WORKER_SLUG: ClassVar[str] leaf_slug: ClassVar[str] @@ -138,4 +148,5 @@ def _spec_for(self, slug, deps, desc): description=desc, assigned_worker_slug=AssignedWorkerSlug(worker_slug), depends_on=[TaskSlug(d) for d in deps], + evaluators=smoke_evaluators_for_slug(slug), ) diff --git a/tests/fixtures/smoke_components/smoke_base/sadpath.py b/tests/fixtures/smoke_components/smoke_base/sadpath.py index abae4689d..3d743b965 100644 --- a/tests/fixtures/smoke_components/smoke_base/sadpath.py +++ b/tests/fixtures/smoke_components/smoke_base/sadpath.py @@ -1,8 +1,9 @@ """Shared smoke sad-path helpers. -The canonical sad path routes ``l_2`` to a failing leaf. ``l_3`` depends -on ``l_2``, so runtime propagation should leave ``l_3`` blocked and never -started while independent branches continue normally. +The canonical sad path routes ``environment-probe`` to a failing leaf. +``metadata-validate`` depends on it, so runtime propagation should leave +``metadata-validate`` blocked and never started while independent branches +continue normally. """ from typing import ClassVar @@ -10,7 +11,11 @@ from e2b_code_interpreter import AsyncSandbox # type: ignore[import-untyped] from ergon_core.api import WorkerContext from ergon_core.core.persistence.shared.types import AssignedWorkerSlug, TaskSlug -from tests.fixtures.smoke_components.smoke_base.dynamic_tasks import SmokeChildTaskSpec +from tests.fixtures.smoke_components.smoke_base.constants import ENV_PROBE_SLUG, SMOKE_OUTPUT_DIR +from tests.fixtures.smoke_components.smoke_base.dynamic_tasks import ( + SmokeChildTaskSpec, + smoke_evaluators_for_slug, +) from tests.fixtures.smoke_components.smoke_base.subworker import SubworkerResult @@ -18,7 +23,7 @@ class AlwaysFailSubworker: """Writes partial work and runs a probe before returning failure.""" async def work(self, task_id: str, sandbox: AsyncSandbox) -> SubworkerResult: - partial_path = f"/workspace/final_output/partial_{task_id}.md" + partial_path = f"{SMOKE_OUTPUT_DIR}/partial_{task_id}.md" await sandbox.files.write( partial_path, ( @@ -52,9 +57,9 @@ async def work(self, task_id: str, sandbox: AsyncSandbox) -> SubworkerResult: class SadPathSmokeWorkerMixin: - """Route ``l_2`` to a failing leaf without changing smoke topology.""" + """Route ``environment-probe`` to a failing leaf without changing smoke topology.""" - FAILING_SLUGS: ClassVar[frozenset[str]] = frozenset({"l_2"}) + FAILING_SLUGS: ClassVar[frozenset[str]] = frozenset({ENV_PROBE_SLUG}) FAILING_LEAF_SLUG: ClassVar[str] leaf_slug: ClassVar[str] @@ -65,6 +70,7 @@ def _spec_for(self, slug, deps, desc): description=desc, assigned_worker_slug=AssignedWorkerSlug(leaf_slug), depends_on=[TaskSlug(d) for d in deps], + evaluators=smoke_evaluators_for_slug(slug), ) diff --git a/tests/fixtures/smoke_components/smoke_base/worker_base.py b/tests/fixtures/smoke_components/smoke_base/worker_base.py index 97f1c2bdb..a7ad8db61 100644 --- a/tests/fixtures/smoke_components/smoke_base/worker_base.py +++ b/tests/fixtures/smoke_components/smoke_base/worker_base.py @@ -26,6 +26,7 @@ from tests.fixtures.smoke_components.smoke_base.constants import SUBTASK_GRAPH from tests.fixtures.smoke_components.smoke_base.dynamic_tasks import ( SmokeChildTaskSpec, + smoke_evaluators_for_slug, smoke_task_from_spec, ) from tests.fixtures.smoke_components.smoke_base.metrics import smoke_assistant_chunk @@ -118,7 +119,7 @@ def _spec_for( """Overridable per-slug → (assigned_worker_slug, deps) mapping. Default routes every slug to ``self.leaf_slug``. Sad-path - subclasses override this to route specific slugs (e.g. ``l_2``) + subclasses override this to route specific slugs (e.g. ``environment-probe``) to a failing leaf while keeping the 9-subtask topology identical. ``execute`` stays ``@final`` so topology is never changed; only the leaf binding is. @@ -128,4 +129,5 @@ def _spec_for( description=desc, assigned_worker_slug=AssignedWorkerSlug(self.leaf_slug), depends_on=[TaskSlug(d) for d in deps], + evaluators=smoke_evaluators_for_slug(slug), ) diff --git a/tests/fixtures/smoke_components/workers/minif2f_smoke.py b/tests/fixtures/smoke_components/workers/minif2f_smoke.py index d40009b9e..6a7e9cc70 100644 --- a/tests/fixtures/smoke_components/workers/minif2f_smoke.py +++ b/tests/fixtures/smoke_components/workers/minif2f_smoke.py @@ -11,6 +11,7 @@ from typing import ClassVar from e2b_code_interpreter import AsyncSandbox # type: ignore[import-untyped] +from tests.fixtures.smoke_components.smoke_base.constants import SMOKE_OUTPUT_DIR from tests.fixtures.smoke_components.smoke_base.leaf_base import BaseSmokeLeafWorker from tests.fixtures.smoke_components.smoke_base.recursive import ( RecursiveSmokeWorkerBase, @@ -43,7 +44,7 @@ class MiniF2FSubworker: """Writes a trivial .lean proof + runs ``lean --check`` as the probe.""" async def work(self, task_id: str, sandbox: AsyncSandbox) -> SubworkerResult: - proof_path = f"/workspace/final_output/proof_{task_id}.lean" + proof_path = f"{SMOKE_OUTPUT_DIR}/proof_{task_id}.lean" await sandbox.files.write(proof_path, LEAN_SOURCE) # ``|| true`` keeps the leaf-side probe exit deterministic even if @@ -54,7 +55,7 @@ async def work(self, task_id: str, sandbox: AsyncSandbox) -> SubworkerResult: timeout=60, ) probe_stdout = ("" if probe.stdout is None else probe.stdout).strip()[:4096] - probe_path = f"/workspace/final_output/probe_{task_id}.json" + probe_path = f"{SMOKE_OUTPUT_DIR}/probe_{task_id}.json" await sandbox.files.write( probe_path, json.dumps({"exit_code": probe.exit_code, "stdout": probe_stdout}), @@ -70,11 +71,12 @@ class MiniF2FSmokeLeafWorker(BaseSmokeLeafWorker): """Registered leaf that delegates to ``MiniF2FSubworker``.""" type_slug: ClassVar[str] = "minif2f-smoke-leaf" + toolkit_type: ClassVar[str] = "ergon_builtins.benchmarks.minif2f.toolkit:MiniF2FToolkit" subworker_cls: ClassVar[type] = MiniF2FSubworker class MiniF2FRecursiveSmokeWorker(RecursiveSmokeWorkerBase): - """Nested ``l_2`` worker that delegates nested leaves to MiniF2F.""" + """Nested ``environment-probe`` worker that delegates nested leaves to MiniF2F.""" type_slug: ClassVar[str] = "minif2f-smoke-recursive-worker" leaf_slug: ClassVar[str] = "minif2f-smoke-leaf" @@ -88,7 +90,7 @@ class MiniF2FFailingLeafWorker(FailingSmokeLeafMixin, BaseSmokeLeafWorker): class MiniF2FSadPathSmokeWorker(SadPathSmokeWorkerMixin, SmokeWorkerBase): - """Parent that routes ``l_2`` to the failing leaf.""" + """Parent that routes ``environment-probe`` to the failing leaf.""" type_slug: ClassVar[str] = "minif2f-sadpath-smoke-worker" leaf_slug: ClassVar[str] = "minif2f-smoke-leaf" diff --git a/tests/fixtures/smoke_components/workers/researchrubrics_smoke.py b/tests/fixtures/smoke_components/workers/researchrubrics_smoke.py index ec8c3453c..aeacac953 100644 --- a/tests/fixtures/smoke_components/workers/researchrubrics_smoke.py +++ b/tests/fixtures/smoke_components/workers/researchrubrics_smoke.py @@ -18,6 +18,7 @@ from typing import ClassVar from e2b_code_interpreter import AsyncSandbox # type: ignore[import-untyped] +from tests.fixtures.smoke_components.smoke_base.constants import SMOKE_OUTPUT_DIR from tests.fixtures.smoke_components.smoke_base.leaf_base import BaseSmokeLeafWorker from tests.fixtures.smoke_components.smoke_base.recursive import ( RecursiveSmokeWorkerBase, @@ -43,7 +44,7 @@ class ResearchRubricsSmokeWorker(RecursiveSmokeWorkerMixin, SmokeWorkerBase): class ResearchRubricsSubworker: """Writes a deterministic markdown report + runs ``wc -l`` as the probe. - Artifacts written to ``/workspace/final_output/`` so the runtime's + Artifacts written to the smoke output directory so the runtime's persist step produces SampleResource rows: - ``report_.md`` — markdown content (content check target) @@ -51,7 +52,7 @@ class ResearchRubricsSubworker: """ async def work(self, task_id: str, sandbox: AsyncSandbox) -> SubworkerResult: - report_path = f"/workspace/final_output/report_{task_id}.md" + report_path = f"{SMOKE_OUTPUT_DIR}/report_{task_id}.md" contents = ( f"# Research report {task_id}\n\n" "Deterministic smoke output. Non-empty body required by criterion.\n" @@ -60,7 +61,7 @@ async def work(self, task_id: str, sandbox: AsyncSandbox) -> SubworkerResult: probe = await sandbox.commands.run(f"wc -l {report_path}", timeout=10) probe_stdout = ("" if probe.stdout is None else probe.stdout).strip() - probe_path = f"/workspace/final_output/probe_{task_id}.json" + probe_path = f"{SMOKE_OUTPUT_DIR}/probe_{task_id}.json" await sandbox.files.write( probe_path, json.dumps({"exit_code": probe.exit_code, "stdout": probe_stdout}), @@ -76,11 +77,14 @@ class ResearchRubricsSmokeLeafWorker(BaseSmokeLeafWorker): """Registered leaf that delegates to ``ResearchRubricsSubworker``.""" type_slug: ClassVar[str] = "researchrubrics-smoke-leaf" + toolkit_type: ClassVar[str] = ( + "ergon_builtins.benchmarks.researchrubrics.toolkit:ResearchRubricsToolkit" + ) subworker_cls: ClassVar[type] = ResearchRubricsSubworker class ResearchRubricsRecursiveSmokeWorker(RecursiveSmokeWorkerBase): - """Nested ``l_2`` worker that delegates nested leaves to ResearchRubrics.""" + """Nested ``environment-probe`` worker that delegates nested leaves to ResearchRubrics.""" type_slug: ClassVar[str] = "researchrubrics-smoke-recursive-worker" leaf_slug: ClassVar[str] = "researchrubrics-smoke-leaf" @@ -94,7 +98,7 @@ class ResearchRubricsFailingLeafWorker(FailingSmokeLeafMixin, BaseSmokeLeafWorke class ResearchRubricsSadPathSmokeWorker(SadPathSmokeWorkerMixin, SmokeWorkerBase): - """Parent that routes ``l_2`` to the failing leaf.""" + """Parent that routes ``environment-probe`` to the failing leaf.""" type_slug: ClassVar[str] = "researchrubrics-sadpath-smoke-worker" leaf_slug: ClassVar[str] = "researchrubrics-smoke-leaf" diff --git a/tests/fixtures/smoke_components/workers/swebench_smoke.py b/tests/fixtures/smoke_components/workers/swebench_smoke.py index 7aa676513..a3cf5b9ba 100644 --- a/tests/fixtures/smoke_components/workers/swebench_smoke.py +++ b/tests/fixtures/smoke_components/workers/swebench_smoke.py @@ -9,6 +9,7 @@ from typing import ClassVar from e2b_code_interpreter import AsyncSandbox # type: ignore[import-untyped] +from tests.fixtures.smoke_components.smoke_base.constants import SMOKE_OUTPUT_DIR from tests.fixtures.smoke_components.smoke_base.leaf_base import BaseSmokeLeafWorker from tests.fixtures.smoke_components.smoke_base.recursive import ( RecursiveSmokeWorkerBase, @@ -44,7 +45,7 @@ class SweBenchSubworker: """Writes a trivial .py file + compiles + executes as the probe.""" async def work(self, task_id: str, sandbox: AsyncSandbox) -> SubworkerResult: - patch_path = f"/workspace/final_output/patch_{task_id}.py" + patch_path = f"{SMOKE_OUTPUT_DIR}/patch_{task_id}.py" await sandbox.files.write(patch_path, PY_SOURCE) probe = await sandbox.commands.run( @@ -52,7 +53,7 @@ async def work(self, task_id: str, sandbox: AsyncSandbox) -> SubworkerResult: timeout=20, ) probe_stdout = ("" if probe.stdout is None else probe.stdout).strip()[:4096] - probe_path = f"/workspace/final_output/probe_{task_id}.json" + probe_path = f"{SMOKE_OUTPUT_DIR}/probe_{task_id}.json" await sandbox.files.write( probe_path, json.dumps({"exit_code": probe.exit_code, "stdout": probe_stdout}), @@ -68,11 +69,14 @@ class SweBenchSmokeLeafWorker(BaseSmokeLeafWorker): """Registered leaf that delegates to ``SweBenchSubworker``.""" type_slug: ClassVar[str] = "swebench-smoke-leaf" + toolkit_type: ClassVar[str] = ( + "ergon_builtins.benchmarks.swebench_verified.toolkit:SWEBenchToolkit" + ) subworker_cls: ClassVar[type] = SweBenchSubworker class SweBenchRecursiveSmokeWorker(RecursiveSmokeWorkerBase): - """Nested ``l_2`` worker that delegates nested leaves to SWE-Bench.""" + """Nested ``environment-probe`` worker that delegates nested leaves to SWE-Bench.""" type_slug: ClassVar[str] = "swebench-smoke-recursive-worker" leaf_slug: ClassVar[str] = "swebench-smoke-leaf" @@ -86,7 +90,7 @@ class SweBenchFailingLeafWorker(FailingSmokeLeafMixin, BaseSmokeLeafWorker): class SweBenchSadPathSmokeWorker(SadPathSmokeWorkerMixin, SmokeWorkerBase): - """Parent that routes ``l_2`` to the failing leaf.""" + """Parent that routes ``environment-probe`` to the failing leaf.""" type_slug: ClassVar[str] = "swebench-sadpath-smoke-worker" leaf_slug: ClassVar[str] = "swebench-smoke-leaf" diff --git a/tests/unit/smoke_base/test_smoke_real_sandbox_posture.py b/tests/unit/smoke_base/test_smoke_real_sandbox_posture.py new file mode 100644 index 000000000..b36325cea --- /dev/null +++ b/tests/unit/smoke_base/test_smoke_real_sandbox_posture.py @@ -0,0 +1,205 @@ +"""Static posture contracts for canonical smoke fixtures.""" + +from copy import deepcopy + +import pytest + +from ergon_core.api import Task +from ergon_core.core.persistence.shared.types import AssignedWorkerSlug, TaskSlug +from tests.fixtures.smoke_components.benchmarks import ( + GDPEvalSmokeEnvironment, + MiniF2FSmokeEnvironment, + ResearchRubricsSmokeEnvironment, + SweBenchSmokeEnvironment, +) +from tests.fixtures.smoke_components.smoke_base.constants import ( + ENV_PROBE_SLUG, + EXPECTED_SUBTASK_SLUGS, + EXPECTED_RESOURCE_HANDOFF, + HANDOFF_RESOURCE_NAME, + HANDOFF_VERIFY_SLUG, + NESTED_LINE_SLUGS, + SMOKE_THREAD_TOPIC, + SOURCE_REVIEW_SLUG, + SUBTASK_GRAPH, +) +from tests.fixtures.smoke_components.smoke_base.dynamic_tasks import ( + SmokeChildTaskSpec, + smoke_task_from_spec, +) +from tests.fixtures.smoke_components.smoke_base.recursive import ( + NESTED_SUBTASK_GRAPH, +) + + +EXPECTED_ROOT_SANDBOX_TYPES = { + ResearchRubricsSmokeEnvironment: ( + "ergon_builtins.benchmarks.researchrubrics.sandbox:ResearchE2BSandbox" + ), + MiniF2FSmokeEnvironment: "ergon_builtins.benchmarks.minif2f.sandbox:LeanSandbox", + SweBenchSmokeEnvironment: ( + "ergon_builtins.benchmarks.swebench_verified.sandbox:SWEBenchSandbox" + ), + GDPEvalSmokeEnvironment: "ergon_builtins.benchmarks.gdpeval.sandbox:GDPEvalSandbox", +} + +EXPECTED_ROOT_POSTURE = { + ResearchRubricsSmokeEnvironment: ( + "smoke-001", + "Review the supplied research notes and produce a concise evidence-backed report.", + ), + MiniF2FSmokeEnvironment: ( + "mathd_algebra_478", + "Verify the supplied Lean theorem and record proof evidence.", + ), + SweBenchSmokeEnvironment: ( + "astropy__astropy-12907", + "Inspect the Python issue and produce a minimal source patch with evidence.", + ), + GDPEvalSmokeEnvironment: ( + "gdpeval-smoke-001", + "Process the reference documents and write a structured output bundle.", + ), +} + +EXPECTED_CHILD_SLUGS = ( + "source-review", + "handoff-verify", + "primary-artifact", + "artifact-summary", + "metadata-review", + "environment-probe", + "metadata-validate", + "evidence-artifact", + "completion-marker", +) + +EXPECTED_CHILD_DEPENDENCIES = { + "source-review": (), + "handoff-verify": ("source-review",), + "primary-artifact": ("source-review",), + "artifact-summary": ("handoff-verify", "primary-artifact"), + "metadata-review": (), + "environment-probe": ("metadata-review",), + "metadata-validate": ("environment-probe",), + "evidence-artifact": (), + "completion-marker": (), +} + +EXPECTED_NESTED_SLUGS = ("nested-input-review", "nested-verification") + + +def _root_task(environment_cls: type) -> Task: + environment = environment_cls() + sample = next(iter(environment.iter_samples())) + [task] = sample.tasks + return task + + +def _sandbox_type(task: Task) -> str: + return task.model_dump(mode="json")["sandbox"]["_type"] + + +def _child_task(parent_task: Task) -> Task: + return smoke_task_from_spec( + parent_task=parent_task, + spec=SmokeChildTaskSpec( + task_slug=TaskSlug("child"), + description="child task", + assigned_worker_slug=AssignedWorkerSlug("researchrubrics-smoke-leaf"), + depends_on=[], + ), + model="openai:gpt-4o", + ) + + +def test_smoke_root_tasks_use_real_environment_sandboxes() -> None: + for environment_cls, expected_type in EXPECTED_ROOT_SANDBOX_TYPES.items(): + task = _root_task(environment_cls) + assert _sandbox_type(task) == expected_type + assert "SmokePublicSandbox" not in expected_type + assert "TestSandbox" not in expected_type + + +def test_smoke_dynamic_tasks_inherit_deepcopy_of_parent_sandbox() -> None: + for environment_cls, expected_type in EXPECTED_ROOT_SANDBOX_TYPES.items(): + parent_task = _root_task(environment_cls) + original_sandbox = parent_task.sandbox + child_task = _child_task(parent_task) + + assert _sandbox_type(child_task) == expected_type + assert child_task.sandbox == original_sandbox + assert child_task.sandbox is not original_sandbox + + +def test_smoke_dynamic_task_requires_parent_sandbox() -> None: + parent_task = deepcopy(_root_task(ResearchRubricsSmokeEnvironment)) + parent_task.sandbox = None + + with pytest.raises(ValueError, match="parent task must define a sandbox"): + _child_task(parent_task) + + +def test_smoke_roots_have_semantic_slugs_and_descriptions() -> None: + for environment_cls, (expected_slug, expected_description) in EXPECTED_ROOT_POSTURE.items(): + task = _root_task(environment_cls) + assert task.task_slug == expected_slug + assert task.description == expected_description + + +def test_smoke_child_topology_uses_semantic_slugs_and_descriptions() -> None: + assert EXPECTED_SUBTASK_SLUGS == EXPECTED_CHILD_SLUGS + + graph_by_slug = {slug: (deps, description) for slug, deps, description in SUBTASK_GRAPH} + assert tuple(graph_by_slug) == EXPECTED_CHILD_SLUGS + + for slug, expected_deps in EXPECTED_CHILD_DEPENDENCIES.items(): + deps, description = graph_by_slug[slug] + assert deps == expected_deps + assert description + assert "Diamond" not in description + assert "Line node" not in description + + +def test_smoke_nested_topology_uses_semantic_slugs() -> None: + assert NESTED_LINE_SLUGS == EXPECTED_NESTED_SLUGS + assert tuple(slug for slug, _deps, _description in NESTED_SUBTASK_GRAPH) == ( + EXPECTED_NESTED_SLUGS + ) + assert NESTED_SUBTASK_GRAPH[1][1] == (EXPECTED_NESTED_SLUGS[0],) + + +def test_smoke_handoff_constants_point_to_semantic_tasks() -> None: + assert EXPECTED_RESOURCE_HANDOFF.producer_slug == SOURCE_REVIEW_SLUG + assert EXPECTED_RESOURCE_HANDOFF.consumer_slug == HANDOFF_VERIFY_SLUG + assert EXPECTED_RESOURCE_HANDOFF.resource_name == HANDOFF_RESOURCE_NAME + assert SMOKE_THREAD_TOPIC == "smoke-coordination" + + +def test_recursive_worker_routes_the_semantic_probe_node() -> None: + from tests.fixtures.smoke_components.smoke_base.recursive import RecursiveSmokeWorkerMixin + + assert RecursiveSmokeWorkerMixin.RECURSIVE_SLUGS == frozenset({ENV_PROBE_SLUG}) + + +def test_smoke_worker_toolkit_binding_constants_when_available() -> None: + bindings = { + "tests.fixtures.smoke_components.workers.researchrubrics_smoke": ( + "ResearchRubricsSmokeLeafWorker", + "ergon_builtins.benchmarks.researchrubrics.toolkit:ResearchRubricsToolkit", + ), + "tests.fixtures.smoke_components.workers.minif2f_smoke": ( + "MiniF2FSmokeLeafWorker", + "ergon_builtins.benchmarks.minif2f.toolkit:MiniF2FToolkit", + ), + "tests.fixtures.smoke_components.workers.swebench_smoke": ( + "SweBenchSmokeLeafWorker", + "ergon_builtins.benchmarks.swebench_verified.toolkit:SWEBenchToolkit", + ), + } + + for module_name, (class_name, expected_type) in bindings.items(): + module = __import__(module_name, fromlist=[class_name]) + worker_cls = getattr(module, class_name) + if hasattr(worker_cls, "toolkit_type"): + assert worker_cls.toolkit_type == expected_type