feat(checker): persist verification findings across turns

This commit is contained in:
somew1nd committed 2026-09-01 20:57:10 -07:00
1 parent 98ee55f5d2
commit 3145de181b
13 files changed
+588 -64

No files matched your search

+26
View File
@@ -84,3 +84,29 @@ async def test_planner_initial_plan(mock_context):
assert isinstance(human_msg_content, list)
assert human_msg_content[0]["type"] == "text"
assert "Goal: Test Goal" in human_msg_content[0]["text"]
def test_validate_plan_format_single_sourced_status_alphabet():
"""Regression: the Planner's format gate is single-sourced from
plan_grammar — a legal in-progress '[/]' line must never be rejected."""
from artemis.agents.planner.planner import validate_plan_format
valid, err = validate_plan_format(
"- [x] Done milestone\n- [/] In-progress milestone\n - [ ] Sub\n- [!] Blocked\n"
)
assert valid is True and err == ""
# Every grammar status char is accepted; unknown ones are still rejected.
from artemis.utils.plan_grammar import STATUS_CHARS
for c in STATUS_CHARS:
ok, _ = validate_plan_format(f"- [{c}] Milestone\n")
assert ok is True
bad, message = validate_plan_format("- [z] Bogus status\n")
assert bad is False
assert "Invalid task status" in message
empty, message = validate_plan_format("# just prose\n")
assert empty is False
assert "at least one subgoal" in message
+150 -1
View File
@@ -27,13 +27,15 @@ from artemis.context import ArtemisContext, ExecutionSetup
from artemis.graph.checkpoints import (
CheckpointRun,
PendingCheckpoint,
checker_note_key,
harvest_finished_checkpoints,
harvest_run,
is_checker_note_key,
queue_checkpoints,
read_ledger,
spawn_pending_checkpoints,
)
from artemis.graph.graph import execution_check_node, wrap_note_tool
from artemis.graph.graph import execution_check_node, wrap_note_tool, wrap_update_note_tool
from artemis.graph.state import State
from artemis.utils.plan_grammar import CheckItem, parse_plan, subgoal_hash
@@ -50,6 +52,9 @@ PLAN_WITH_CHECKS = (
def _make_ctx(tmp_path, **setup_kwargs):
setup_kwargs.setdefault("disable_checker", False)
# These tests exercise the midway checkpoint machinery, which is off in
# the factory layering — pin it on unless a test overrides it.
setup_kwargs.setdefault("disable_midway_checks", False)
ctx = MagicMock(spec=ArtemisContext)
ctx.execution_setup = ExecutionSetup(**setup_kwargs)
ctx.data_engine = MagicMock()
@@ -528,6 +533,150 @@ async def test_timeout_records_inconclusive_not_passed(tmp_path):
assert all(r["status"] != "passed" for r in records)
# --- Verify-finding four-layer persistence -------------------------------------------
def _checker_note_path(tmp_path):
return tmp_path / "notes" / f"{checker_note_key(GOAL_KEY)}.md"
@pytest.mark.asyncio
async def test_verify_fail_writes_finding_line_and_checker_note(tmp_path):
plan_path = _write_plan(tmp_path)
ctx = _make_ctx(tmp_path)
run = _done_run(ctx, CheckReport(verdicts=[_verdict("verify", "failed")]))
harvest_run(ctx, _make_state(), run, allow_side_effects=True)
content = plan_path.read_text(encoding="utf-8")
lines = content.splitlines()
# Standing headline pinned directly under the (now reverted) subgoal,
# naming the failed criterion and pointing at the detail note.
idx = lines.index(f"- [/] {GOAL_TEXT}")
assert lines[idx + 1].startswith(" - finding: verify failed")
assert checker_note_key(GOAL_KEY) in lines[idx + 1]
# Layer 3: the system note carries the full criterion/evidence/suggestion.
note = _checker_note_path(tmp_path).read_text(encoding="utf-8")
assert "the alarm list shows 7:30 AM" in note
assert "concrete evidence" in note
assert "fix it" in note
@pytest.mark.asyncio
async def test_finding_line_regrows_on_model_plan_write(tmp_path):
plan_path = _write_plan(tmp_path)
ctx = _make_ctx(tmp_path)
run = _done_run(ctx, CheckReport(verdicts=[_verdict("verify", "failed")]))
harvest_run(ctx, _make_state(), run, allow_side_effects=True)
assert "- finding:" in plan_path.read_text(encoding="utf-8")
# The model rewrites the plan WITHOUT the finding line...
clean_rewrite = PLAN_WITH_CHECKS.replace(f"- [x] {GOAL_TEXT}", f"- [/] {GOAL_TEXT}")
async def fake_invoke(tool, args, tool_call_id, state, record_trace=None):
plan_path.write_text(args["content"], encoding="utf-8")
return "Success"
mock_tool = MagicMock()
mock_tool.name = "save_note"
mock_tool.description = "save"
with patch("artemis.graph.graph.invoke_tool_with_injection", side_effect=fake_invoke):
wrapped = wrap_note_tool(ctx, mock_tool)
await wrapped.ainvoke(
{"key": "task_plan", "content": clean_rewrite, "tool_call_id": "t1"}
)
# ...and the deterministic projection grows it back on the write path.
assert "- finding: verify failed" in plan_path.read_text(encoding="utf-8")
@pytest.mark.asyncio
async def test_finding_line_removed_after_verify_passes_note_kept(tmp_path):
plan_path = _write_plan(tmp_path)
ctx = _make_ctx(tmp_path)
run1 = _done_run(ctx, CheckReport(verdicts=[_verdict("verify", "failed")]))
harvest_run(ctx, _make_state(), run1, allow_side_effects=True)
assert "- finding:" in plan_path.read_text(encoding="utf-8")
run2 = _done_run(
ctx,
CheckReport(verdicts=[_verdict("verify", "passed")]),
attempt_id=f"{GOAL_KEY}#2",
)
harvest_run(ctx, _make_state(), run2, allow_side_effects=True)
# Layer 2 headline retired; layer 3 note stays on disk with the full
# per-attempt repair log (fail then pass).
assert "- finding:" not in plan_path.read_text(encoding="utf-8")
note = _checker_note_path(tmp_path).read_text(encoding="utf-8")
assert f"attempt {GOAL_KEY}#1 — verify failed" in note
assert f"attempt {GOAL_KEY}#2 — verify passed" in note
@pytest.mark.asyncio
async def test_finding_line_removed_on_repair_quota_exhaustion(tmp_path):
plan_path = _write_plan(tmp_path)
ctx = _make_ctx(tmp_path, checkpoint_max_repairs=1)
run1 = _done_run(ctx, CheckReport(verdicts=[_verdict("verify", "failed")]))
harvest_run(ctx, _make_state(), run1, allow_side_effects=True)
assert "- finding:" in plan_path.read_text(encoding="utf-8")
# Re-complete, fail again: quota (1) is exhausted -> finding settled.
plan_path.write_text(
plan_path.read_text(encoding="utf-8").replace(
f"- [/] {GOAL_TEXT}", f"- [x] {GOAL_TEXT}"
),
encoding="utf-8",
)
run2 = _done_run(
ctx,
CheckReport(verdicts=[_verdict("verify", "failed")]),
attempt_id=f"{GOAL_KEY}#2",
)
harvest_run(ctx, _make_state(), run2, allow_side_effects=True)
content = plan_path.read_text(encoding="utf-8")
assert "- finding:" not in content
assert f"- [x] {GOAL_TEXT}" in content # no further revert either
@pytest.mark.asyncio
async def test_checker_note_prefix_writes_rejected_by_wrappers(tmp_path):
ctx = _make_ctx(tmp_path)
save_tool = MagicMock()
save_tool.name = "save_note"
save_tool.description = "save"
update_tool = MagicMock()
update_tool.name = "update_note"
update_tool.description = "update"
with patch(
"artemis.graph.graph.invoke_tool_with_injection", new_callable=AsyncMock
) as inv:
wrapped_save = wrap_note_tool(ctx, save_tool)
result = await wrapped_save.ainvoke(
{"key": checker_note_key(GOAL_KEY), "content": "spoof", "tool_call_id": "t1"}
)
assert "reserved" in str(result.content)
wrapped_update = wrap_update_note_tool(ctx, update_tool)
result2 = await wrapped_update.ainvoke(
{
"key": "checker:abc",
"target": "a",
"replacement": "b",
"tool_call_id": "t2",
}
)
assert "reserved" in str(result2.content)
inv.assert_not_awaited()
assert is_checker_note_key("checker-123") and is_checker_note_key("checker:123")
assert not is_checker_note_key("checkerboard_notes") and not is_checker_note_key("notes")
@pytest.mark.asyncio
async def test_exception_fail_open_releases_but_books_inconclusive(tmp_path):
plan_path = _write_plan(tmp_path)
+51 -10
View File
@@ -179,13 +179,13 @@ def test_agent_config_loading():
assert agent_cfg.explorer.caching is True
assert "explorer" in agent_cfg.denylisted_tools
assert agent_cfg.video_analyzer.enable_ledger is True
assert agent_cfg.planner_validation.enabled is False
assert agent_cfg.planner_validation.enabled is True
assert agent_cfg.planner_validation.similarity_threshold == 0.85
assert agent_cfg.committee.enabled is False
assert agent_cfg.committee.debate_rounds == 2
assert agent_cfg.checker.enabled is False
assert agent_cfg.checker.enabled is True
assert agent_cfg.checker.max_iterations == 20
assert agent_cfg.checker.midway_checks is True
assert agent_cfg.checker.midway_checks is False
assert agent_cfg.checker.final_check is True
assert agent_cfg.checker.assert_failure_policy == "continue"
assert agent_cfg.checker.device_probes is True
@@ -194,9 +194,9 @@ def test_agent_config_loading():
assert agent_cfg.flash.max_turns == 30
assert agent_cfg.flash.explorer_mode == "flash"
assert agent_cfg.pro.explorer.mode == "flash"
assert agent_cfg.pro.checker.enabled is False
assert agent_cfg.pro.checker.enabled is True
assert agent_cfg.pro.committee.enabled is False
assert agent_cfg.pro.planner_validation.enabled is False
assert agent_cfg.pro.planner_validation.enabled is True
assert agent_cfg.pro.video_analyzer.enable_ledger is True
@@ -255,10 +255,10 @@ def test_planner_validation_builder_and_milestones():
from artemis.sdk.builders.agent_config_builder import AgentConfigBuilder
from artemis.utils.plan_grammar import milestones_changed, parse_plan
# Default builder inherits from artemis.jsonc (enabled=False, similarity_threshold=0.85)
# Default builder inherits from artemis.jsonc (enabled=True, similarity_threshold=0.85)
builder = AgentConfigBuilder()
cfg = builder.build()
assert cfg.disable_planner_validation is True
assert cfg.disable_planner_validation is False
assert cfg.planner_validation_threshold == 0.85
# Fluent enabling
@@ -338,12 +338,12 @@ def test_checker_builder_and_context_propagation():
from artemis.sdk.agent import Agent
from artemis.sdk.builders.agent_config_builder import AgentConfigBuilder
# Default builder inherits from artemis.jsonc (enabled=False, max_iterations=20)
# Default builder inherits from artemis.jsonc (enabled=True, midway off, final on)
builder = AgentConfigBuilder()
cfg = builder.build()
assert cfg.disable_checker is True
assert cfg.disable_checker is False
assert cfg.checker_max_iterations == 20
assert cfg.disable_midway_checks is False
assert cfg.disable_midway_checks is True
assert cfg.disable_final_check is False
# Fluent enabling with the new two-gate knobs
@@ -402,6 +402,47 @@ def test_checker_builder_and_context_propagation():
assert ctx.execution_setup.final_check_enabled is False
def test_factory_default_verification_layering():
"""Contract: out of the box, the verification stack is layered as
final check ON / planner validation (ratchet) ON / midway checks OFF."""
from artemis.config.agent import AgentGlobalConfig, CheckerConfig, PlannerValidationConfig
from artemis.context import ExecutionSetup
from artemis.sdk.builders.agent_config_builder import AgentConfigBuilder
# Config-model factory defaults
assert CheckerConfig().enabled is True
assert CheckerConfig().midway_checks is False
assert CheckerConfig().final_check is True
assert PlannerValidationConfig().enabled is True
# Bare ExecutionSetup carries the same layering, including the effective
# gate semantics (master on + midway off => final review runs, no midway
# checkpoint ever spawns).
setup = ExecutionSetup()
assert setup.disable_checker is False
assert setup.disable_midway_checks is True
assert setup.disable_final_check is False
assert setup.disable_planner_validation is False
assert setup.final_check_enabled is True
assert setup.midway_checks_enabled is False
assert setup.checks_enabled is True
# Builder default (fed by config/artemis.jsonc) agrees
cfg = AgentConfigBuilder().build()
assert cfg.disable_checker is False
assert cfg.disable_midway_checks is True
assert cfg.disable_final_check is False
assert cfg.disable_planner_validation is False
# Shipped artemis.jsonc agrees with the model factory defaults
agent_cfg = load_agent_config()
assert agent_cfg.checker.enabled is True
assert agent_cfg.checker.midway_checks is False
assert agent_cfg.checker.final_check is True
assert agent_cfg.planner_validation.enabled is True
assert AgentGlobalConfig().checker.enabled is True
def test_explorer_builder_and_resolution(monkeypatch):
"""Test AgentConfigBuilder explorer methods and multi-tier resolution logic."""
from artemis.context import ArtemisContext, DeviceContext, DevicePlatform
+56
View File
@@ -239,3 +239,59 @@ def test_prompts_carry_the_single_sourced_grammar():
planner = json.loads(planner_raw)
guidelines = planner["blocks"]["validator_guidelines"]
assert "until manually stopped" not in guidelines
# --- Finding lines (system channel; single-direction projection) ---------------------
def test_finding_lines_are_invisible_to_the_machine_channel():
from artemis.utils.plan_grammar import apply_finding_lines
plan = "- [x] G1\n - verify: V1\n- [ ] G2\n"
key = subgoal_hash("G1")
with_finding = apply_finding_lines(
plan, {key: "verify failed — 'V1' (details: note 'checker-x')"}
)
assert "- finding: verify failed" in with_finding
before, after = parse_plan(plan), parse_plan(with_finding)
# Same items, same hashes, same check items: no milestone drift, no
# planner-validation trigger, no check multiset change.
assert [i.key for i in after.items] == [i.key for i in before.items]
assert milestones_changed(before, after) is False
assert check_items_changed(before, after) is False
assert new_top_level_completions(before, after) == []
assert unintended_milestone_edits(before, after) == []
def test_apply_finding_lines_projection_is_idempotent_and_anchored():
from artemis.utils.plan_grammar import apply_finding_lines, strip_finding_lines
plan = "- [x] G1\n - verify: V1\n- [/] G2\n"
findings = {subgoal_hash("G1"): "headline one"}
once = apply_finding_lines(plan, findings)
twice = apply_finding_lines(once, findings)
assert once == twice # idempotent projection
# Anchored directly under its parent subgoal
lines = once.splitlines()
assert lines[lines.index("- [x] G1") + 1] == " - finding: headline one"
# Resolution: an empty registry strips the line back out
assert apply_finding_lines(once, {}) == plan
assert strip_finding_lines(once) == plan
def test_apply_finding_lines_drops_entries_for_missing_subgoals():
from artemis.utils.plan_grammar import apply_finding_lines
plan = "- [x] G1 (rephrased)\n"
projected = apply_finding_lines(plan, {subgoal_hash("G1"): "stale headline"})
assert "finding" not in projected
def test_finding_grammar_doc_is_conditional():
base = render_plan_grammar_spec(include_checks=False)
extended = render_plan_grammar_spec(include_checks=True)
assert "finding" not in base
assert "- finding:" in extended and "checker-" in extended