Lap Tran
bc7daf8590
- **Restore 82 completed tasks** from tasks/complete/ back to tasks/ top level (all <7 days old per the cleanup policy; premature bulk archive was fixed). - **Dashboard: fix scroll-reset on auto-refresh** — renderBoard rebuilds the board via innerHTML every 2s, destroying each column-body's scrollTop. Now snapshots column-body scrollTop + board.scrollLeft + view.scrollTop before rebuild and restores after (matched by PHASE_GROUPS index). - **Dashboard UI additions** (pre-existing unstaged work): approval section cards, transition buttons, inline artifact editor (textarea for writing missing SPEC/VERDICT/etc from the detail modal). - **Bind ornith as Implement model** — config.md: Model explicit to omlx/Ornith-1.0-35B-4bit-mlx, context window 32768. Interactive autopilot already used ornith via opencode default; now explicit. - **Fix cleanup stub** — automaton-cleanup.sh had a stale --project arg pointing at a pytest temp dir (test isolation leak). Rewired to point at ~/.automaton. - **Fix plist-isolation test** — test asserted host plist doesn't exist, but a real install creates it. Now snapshots mtime before run, asserts unchanged after (only a write during the test counts as bleed). - **New Playwright smoke test** (tests/test_dashboard_ui.py) — 2 tests: board renders tasks, column scroll survives auto-refresh tick. Verified the test fails without the scroll fix (scrollTop resets to 0). Skipped via importorskip when playwright is absent (main CI stays green). - **Clarify SI loop scope in README** — new-project onboarding section documents the framework-scoped self-improvement loop and options (leave/pause/create project loop). - **CHANGELOG** documents all changes including the known model-divergence gap (mde tasks marked complete but per-role model binding was never implemented).
411 lines
16 KiB
Python
411 lines
16 KiB
Python
"""Tests for automaton.dashboard.core.task."""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from automaton.dashboard.core.task import (
|
|
determine_task_state,
|
|
discover_tasks,
|
|
parse_sub_tasks,
|
|
parse_waves,
|
|
parse_vram_config,
|
|
WaveGroup,
|
|
TaskState,
|
|
)
|
|
|
|
|
|
def _make_task(tmp_path: Path, name: str, artifacts: dict[str, str]) -> Path:
|
|
"""Create a task folder with the given artifact contents."""
|
|
task_dir = tmp_path / name
|
|
task_dir.mkdir(parents=True)
|
|
for filename, content in artifacts.items():
|
|
(task_dir / filename).write_text(content)
|
|
return task_dir
|
|
|
|
|
|
def test_backlog_state(tmp_path: Path) -> None:
|
|
task_dir = _make_task(tmp_path, "backlog-task", {})
|
|
state, phase_raw, artifacts = determine_task_state(task_dir)
|
|
assert state == TaskState.BACKLOG
|
|
assert not artifacts
|
|
|
|
|
|
def test_research_state(tmp_path: Path) -> None:
|
|
task_dir = _make_task(tmp_path, "research-task", {"SPEC.md": "# Spec"})
|
|
state, phase_raw, artifacts = determine_task_state(task_dir)
|
|
assert state == TaskState.RESEARCH
|
|
assert "SPEC.md" in artifacts
|
|
|
|
|
|
def test_implementation_state(tmp_path: Path) -> None:
|
|
task_dir = _make_task(
|
|
tmp_path,
|
|
"impl-task",
|
|
{"SPEC.md": "# Spec", "TEST_PLAN.md": "# Tests", "IMPLEMENTATION.md": "# Impl"},
|
|
)
|
|
state, phase_raw, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.IMPLEMENT
|
|
|
|
|
|
def test_implementation_from_test_plan(tmp_path: Path) -> None:
|
|
task_dir = _make_task(
|
|
tmp_path,
|
|
"impl-test-plan",
|
|
{"SPEC.md": "# Spec", "TEST_PLAN.md": "# Tests"},
|
|
)
|
|
state, phase_raw, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.TEST_DESIGN
|
|
|
|
|
|
def test_bug_find_state(tmp_path: Path) -> None:
|
|
task_dir = _make_task(
|
|
tmp_path,
|
|
"bug-task",
|
|
{"SPEC.md": "# Spec", "IMPLEMENTATION.md": "# Impl", "BUG_REPORT.md": "# Bugs"},
|
|
)
|
|
state, phase_raw, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.BUG_FIND
|
|
|
|
|
|
def test_adv_bug_find_state(tmp_path: Path) -> None:
|
|
task_dir = _make_task(
|
|
tmp_path,
|
|
"adv-task",
|
|
{
|
|
"SPEC.md": "# Spec",
|
|
"IMPLEMENTATION.md": "# Impl",
|
|
"BUG_REPORT.md": "# Bugs",
|
|
"ADVERSARIAL_BUG_REPORT.md": "# Adv",
|
|
},
|
|
)
|
|
state, phase_raw, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.ADV_BUG_FIND
|
|
|
|
|
|
def test_done_state(tmp_path: Path) -> None:
|
|
task_dir = _make_task(
|
|
tmp_path,
|
|
"done-task",
|
|
{"SPEC.md": "# Spec", "VERDICT.md": "## Status: PASS"},
|
|
)
|
|
state, phase_raw, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.DONE
|
|
|
|
|
|
def test_blocked_state(tmp_path: Path) -> None:
|
|
task_dir = _make_task(
|
|
tmp_path,
|
|
"blocked-task",
|
|
{"SPEC.md": "# Spec", "VERDICT.md": "## Status: FAIL"},
|
|
)
|
|
state, phase_raw, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.BLOCKED
|
|
|
|
|
|
def test_empty_verdict_is_blocked(tmp_path: Path) -> None:
|
|
task_dir = _make_task(tmp_path, "empty-verdict", {"VERDICT.md": ""})
|
|
state, phase_raw, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.BLOCKED
|
|
|
|
|
|
def test_parse_sub_tasks(tmp_path: Path) -> None:
|
|
parent = tmp_path / "parent"
|
|
parent.mkdir()
|
|
sub = parent / "subtasks" / "sub-a"
|
|
sub.mkdir(parents=True)
|
|
(sub / "SPEC.md").write_text("# Spec")
|
|
(sub / "VERDICT.md").write_text("## Status: PASS")
|
|
|
|
sub_tasks = parse_sub_tasks(parent)
|
|
assert len(sub_tasks) == 1
|
|
assert sub_tasks[0].name == "sub-a"
|
|
assert sub_tasks[0].has_verdict is True
|
|
assert sub_tasks[0].verdict_status == "PASS"
|
|
|
|
|
|
def test_discover_tasks(tmp_path: Path) -> None:
|
|
_make_task(tmp_path, "aaa", {"SPEC.md": "# Spec"})
|
|
_make_task(tmp_path, "zzz", {"VERDICT.md": "## Status: PASS"})
|
|
tasks = discover_tasks(tmp_path)
|
|
assert len(tasks) == 2
|
|
# Done task should be first (sorted by advancement).
|
|
assert tasks[0].state == TaskState.DONE
|
|
assert tasks[1].state == TaskState.RESEARCH
|
|
|
|
|
|
def test_discover_tasks_skips_complete_container(tmp_path: Path) -> None:
|
|
_make_task(tmp_path, "active", {"SPEC.md": "# Spec"})
|
|
complete = tmp_path / "complete"
|
|
complete.mkdir()
|
|
(complete / "done-task").mkdir()
|
|
(complete / "done-task" / "VERDICT.md").write_text("## Status: PASS")
|
|
(complete / "REVIEW.md").write_text("# Review\n- **Status**: approved")
|
|
|
|
tasks = discover_tasks(tmp_path)
|
|
assert len(tasks) == 1
|
|
assert tasks[0].name == "active"
|
|
|
|
|
|
def test_discover_tasks_skips_subtasks_root(tmp_path: Path) -> None:
|
|
parent = tmp_path / "parent"
|
|
parent.mkdir()
|
|
(parent / "SPEC.md").write_text("# Spec")
|
|
(parent / "subtasks" / "sub-a").mkdir(parents=True)
|
|
(parent / "subtasks" / "sub-a" / "SPEC.md").write_text("# Sub")
|
|
|
|
tasks = discover_tasks(tmp_path)
|
|
assert len(tasks) == 1
|
|
assert tasks[0].name == "parent"
|
|
assert len(tasks[0].sub_tasks) == 1
|
|
|
|
|
|
class TestVerdictParsing:
|
|
"""Tests for structured verdict status parsing (R1-R3 of fix-verdict-parsing SPEC)."""
|
|
|
|
def test_pass_verdict_with_fail_in_findings(self, tmp_path: Path) -> None:
|
|
ver = "## Status: PASS\n\nThe previous FAIL finding was resolved."
|
|
task_dir = _make_task(tmp_path, "pass-with-fail", {"SPEC.md": "# Spec", "VERDICT.md": ver})
|
|
state, phase_raw, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.DONE
|
|
|
|
def test_pass_verdict_with_needs_review_in_body(self, tmp_path: Path) -> None:
|
|
ver = "## Status: PASS\n\nNote: NEEDS_REVIEW was discussed but resolved."
|
|
task_dir = _make_task(tmp_path, "pass-with-nr", {"SPEC.md": "# Spec", "VERDICT.md": ver})
|
|
state, phase_raw, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.DONE
|
|
|
|
def test_fail_verdict_structured(self, tmp_path: Path) -> None:
|
|
ver = "## Status: FAIL\n\n2 tests PASS, 1 test FAIL."
|
|
task_dir = _make_task(tmp_path, "fail-mentions-pass", {"SPEC.md": "# Spec", "VERDICT.md": ver})
|
|
state, phase_raw, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.BLOCKED
|
|
|
|
def test_needs_review_verdict_structured(self, tmp_path: Path) -> None:
|
|
ver = "## Status: NEEDS_REVIEW\n\nSome items PASS but need review."
|
|
task_dir = _make_task(tmp_path, "nr-mentions-pass", {"SPEC.md": "# Spec", "VERDICT.md": ver})
|
|
state, phase_raw, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.BLOCKED
|
|
|
|
def test_verdict_with_bold_status(self, tmp_path: Path) -> None:
|
|
ver = "# Verdict\n\n- **Status**: PASS\n- **Timestamp**: 2025-01-01"
|
|
task_dir = _make_task(tmp_path, "bold-status", {"SPEC.md": "# Spec", "VERDICT.md": ver})
|
|
state, phase_raw, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.DONE
|
|
|
|
def test_verdict_no_status_line(self, tmp_path: Path) -> None:
|
|
ver = "# Verdict\nEverything looks good, PASS!"
|
|
task_dir = _make_task(tmp_path, "no-status-line", {"SPEC.md": "# Spec", "VERDICT.md": ver})
|
|
state, phase_raw, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.REFEREE # no structured ## Status: header → ambiguous
|
|
|
|
def test_verdict_no_status_no_keywords(self, tmp_path: Path) -> None:
|
|
ver = "# Verdict\n\nNeeds further discussion."
|
|
task_dir = _make_task(tmp_path, "no-status-no-keywords", {"SPEC.md": "# Spec", "VERDICT.md": ver})
|
|
state, phase_raw, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.REFEREE
|
|
|
|
|
|
class TestStateMachineAlignment:
|
|
"""Tests for state machine alignment with orchestrate.md (R4)."""
|
|
|
|
def test_implementation_alone_shows_implement(self, tmp_path: Path) -> None:
|
|
task_dir = _make_task(tmp_path, "impl-only", {"IMPLEMENTATION.md": "# Impl"})
|
|
state, phase_raw, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.IMPLEMENT
|
|
|
|
def test_bug_report_without_adversarial(self, tmp_path: Path) -> None:
|
|
task_dir = _make_task(
|
|
tmp_path, "bug-only",
|
|
{"SPEC.md": "# Spec", "IMPLEMENTATION.md": "# Impl", "BUG_REPORT.md": "# Bugs"},
|
|
)
|
|
state, phase_raw, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.BUG_FIND
|
|
|
|
def test_both_bug_reports_shows_adv_bug_find(self, tmp_path: Path) -> None:
|
|
task_dir = _make_task(
|
|
tmp_path, "both-bugs",
|
|
{"SPEC.md": "# Spec", "IMPLEMENTATION.md": "# Impl",
|
|
"BUG_REPORT.md": "# Bugs", "ADVERSARIAL_BUG_REPORT.md": "# Adv"},
|
|
)
|
|
state, phase_raw, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.ADV_BUG_FIND
|
|
|
|
def test_adv_bug_report_alone_shows_bug_find(self, tmp_path: Path) -> None:
|
|
task_dir = _make_task(
|
|
tmp_path, "adv-only",
|
|
{"SPEC.md": "# Spec", "IMPLEMENTATION.md": "# Impl",
|
|
"ADVERSARIAL_BUG_REPORT.md": "# Adv bugs only"},
|
|
)
|
|
state, phase_raw, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.BUG_FIND
|
|
|
|
def test_spec_alone_shows_research(self, tmp_path: Path) -> None:
|
|
task_dir = _make_task(tmp_path, "spec-only", {"SPEC.md": "# Spec"})
|
|
state, phase_raw, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.RESEARCH
|
|
|
|
def test_test_plan_shows_test_design(self, tmp_path: Path) -> None:
|
|
task_dir = _make_task(tmp_path, "testplan", {"SPEC.md": "# Spec", "TEST_PLAN.md": "# Tests"})
|
|
state, phase_raw, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.TEST_DESIGN
|
|
|
|
def test_design_with_spec_shows_design(self, tmp_path: Path) -> None:
|
|
task_dir = _make_task(tmp_path, "design-spec", {"SPEC.md": "# Spec", "DESIGN.md": "# Design"})
|
|
state, phase_raw, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.DESIGN
|
|
|
|
def test_doc_review_shows_doc_review(self, tmp_path: Path) -> None:
|
|
task_dir = _make_task(
|
|
tmp_path, "doc-review",
|
|
{"SPEC.md": "# Spec", "IMPLEMENTATION.md": "# Impl",
|
|
"BUG_REPORT.md": "# Bugs", "ADVERSARIAL_BUG_REPORT.md": "# Adv",
|
|
"DOC_REVIEW.md": "# Docs"},
|
|
)
|
|
state, phase_raw, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.DOC_REVIEW
|
|
|
|
|
|
class TestTaskNameValidation:
|
|
"""Tests for filesystem-sourced task name validation."""
|
|
|
|
def test_valid_task_names(self, tmp_path: Path) -> None:
|
|
for name in ["my-task", "task_1", "Task-Name-123"]:
|
|
d = tmp_path / name
|
|
d.mkdir()
|
|
(d / "SPEC.md").write_text("# Spec")
|
|
tasks = discover_tasks(tmp_path)
|
|
assert len(tasks) == 3
|
|
|
|
def test_invalid_task_name_skipped(self, tmp_path: Path) -> None:
|
|
valid = tmp_path / "good-task"
|
|
valid.mkdir()
|
|
(valid / "SPEC.md").write_text("# Spec")
|
|
bad = tmp_path / "task with spaces"
|
|
bad.mkdir()
|
|
(bad / "SPEC.md").write_text("# Bad")
|
|
tasks = discover_tasks(tmp_path)
|
|
assert len(tasks) == 1
|
|
assert tasks[0].name == "good-task"
|
|
|
|
|
|
class TestParseWaves:
|
|
def test_parse_waves_from_template(self) -> None:
|
|
content = "# Decomposition: Test\n\n## Sub-Tasks\n\n### Wave 1 (Parallel)\n- subtask-a: Implement core\n- subtask-b: Implement model\n\n### Wave 2 (Dependent)\n- subtask-c: Implement UI\n- subtask-d: Implement tests\n"
|
|
waves = parse_waves(content)
|
|
assert len(waves) == 2
|
|
assert waves[0].wave_number == 1
|
|
assert waves[0].label == "Parallel"
|
|
assert waves[0].sub_task_names == ["subtask-a", "subtask-b"]
|
|
assert waves[1].wave_number == 2
|
|
assert waves[1].label == "Dependent"
|
|
assert waves[1].sub_task_names == ["subtask-c", "subtask-d"]
|
|
|
|
def test_parse_waves_empty(self) -> None:
|
|
assert parse_waves("") == []
|
|
assert parse_waves(None) == []
|
|
|
|
def test_parse_waves_no_waves(self) -> None:
|
|
content = "# No waves here\nJust text\n"
|
|
assert parse_waves(content) == []
|
|
|
|
def test_parse_waves_colon_format(self) -> None:
|
|
content = "### Wave 1: Setup\n- task-alpha: Do setup\n### Wave 2: Execution\n- task-beta: Do execution\n"
|
|
waves = parse_waves(content)
|
|
assert len(waves) == 2
|
|
assert waves[0].label == "Setup"
|
|
assert waves[0].sub_task_names == ["task-alpha"]
|
|
assert waves[1].label == "Execution"
|
|
assert waves[1].sub_task_names == ["task-beta"]
|
|
|
|
|
|
class TestDecompositionContent:
|
|
def test_task_has_decomposition_content(self, tmp_path: Path) -> None:
|
|
task_dir = tmp_path / "my-task"
|
|
task_dir.mkdir()
|
|
(task_dir / "SPEC.md").write_text("# Spec")
|
|
(task_dir / "DECOMPOSITION.md").write_text("### Wave 1 (Build)\n- sub-a: Build core\n")
|
|
tasks = discover_tasks(tmp_path)
|
|
assert len(tasks) == 1
|
|
assert tasks[0].decomposition_content is not None
|
|
assert "Wave 1" in tasks[0].decomposition_content
|
|
assert len(tasks[0].waves) == 1
|
|
assert tasks[0].waves[0].sub_task_names == ["sub-a"]
|
|
|
|
|
|
class TestParentSpecAndVramConfig:
|
|
def test_parent_spec_content(self, tmp_path: Path) -> None:
|
|
task_dir = tmp_path / "my-task"
|
|
task_dir.mkdir()
|
|
(task_dir / "SPEC.md").write_text("# Spec")
|
|
(task_dir / "PARENT_SPEC.md").write_text("# Parent Context\nDetails here")
|
|
tasks = discover_tasks(tmp_path)
|
|
assert tasks[0].parent_spec_content is not None
|
|
assert "Parent Context" in tasks[0].parent_spec_content
|
|
|
|
def test_vram_config_content(self, tmp_path: Path) -> None:
|
|
task_dir = tmp_path / "my-task"
|
|
task_dir.mkdir()
|
|
(task_dir / "SPEC.md").write_text("# Spec")
|
|
(task_dir / "VRAM_CONFIG.md").write_text("# VRAM\nmodel: llama-3")
|
|
tasks = discover_tasks(tmp_path)
|
|
assert tasks[0].vram_config_content is not None
|
|
assert "llama-3" in tasks[0].vram_config_content
|
|
|
|
def test_subtask_has_parent_spec(self, tmp_path: Path) -> None:
|
|
tasks_dir = tmp_path
|
|
parent = tasks_dir / "parent-task"
|
|
parent.mkdir()
|
|
(parent / "SPEC.md").write_text("# Parent")
|
|
subtasks_dir = parent / "subtasks"
|
|
subtasks_dir.mkdir()
|
|
sub = subtasks_dir / "child-a"
|
|
sub.mkdir()
|
|
(sub / "SPEC.md").write_text("# Child")
|
|
(sub / "PARENT_SPEC.md").write_text("# Parent Spec for child")
|
|
(sub / "VRAM_CONFIG.md").write_text("# VRAM config")
|
|
sub_tasks = parse_sub_tasks(parent)
|
|
assert len(sub_tasks) == 1
|
|
# discover_tasks doesn't recurse into subtasks for content, but parse_sub_tasks returns SubTask objects
|
|
# Verify the files exist
|
|
assert (sub / "PARENT_SPEC.md").exists()
|
|
|
|
|
|
class TestStatusReason:
|
|
def test_done_task_reason(self, tmp_path: Path) -> None:
|
|
task_dir = tmp_path / "done-task"
|
|
task_dir.mkdir()
|
|
(task_dir / "VERDICT.md").write_text("## Status: PASS\nAll good.\n")
|
|
(task_dir / "SPEC.md").write_text("# Spec\n")
|
|
tasks = discover_tasks(tmp_path)
|
|
assert tasks[0].status_reason == "Verdict: PASS"
|
|
|
|
def test_blocked_fail_reason(self, tmp_path: Path) -> None:
|
|
task_dir = tmp_path / "blocked-task"
|
|
task_dir.mkdir()
|
|
(task_dir / "VERDICT.md").write_text("## Status: FAIL\nBroken.\n")
|
|
(task_dir / "SPEC.md").write_text("# Spec\n")
|
|
tasks = discover_tasks(tmp_path)
|
|
assert "FAIL" in tasks[0].status_reason
|
|
|
|
def test_blocked_needs_review_reason(self, tmp_path: Path) -> None:
|
|
task_dir = tmp_path / "review-task"
|
|
task_dir.mkdir()
|
|
(task_dir / "VERDICT.md").write_text("## Status: NEEDS_REVIEW\nUnclear.\n")
|
|
(task_dir / "SPEC.md").write_text("# Spec\n")
|
|
tasks = discover_tasks(tmp_path)
|
|
assert "NEEDS_REVIEW" in tasks[0].status_reason
|
|
|
|
def test_bug_find_implementation_reason(self, tmp_path: Path) -> None:
|
|
task_dir = tmp_path / "impl-task"
|
|
task_dir.mkdir()
|
|
(task_dir / "IMPLEMENTATION.md").write_text("# Implementation\n")
|
|
tasks = discover_tasks(tmp_path)
|
|
assert "Implementation complete" in tasks[0].status_reason
|
|
|
|
def test_backlog_reason(self, tmp_path: Path) -> None:
|
|
task_dir = tmp_path / "empty-task"
|
|
task_dir.mkdir()
|
|
tasks = discover_tasks(tmp_path)
|
|
assert "No artifacts" in tasks[0].status_reason
|