298 lines
11 KiB
Python
298 lines
11 KiB
Python
# Copyright 2026 Google LLC
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from artemis.utils.plan_grammar import (
|
|
CONTINUOUS_LOOP_TAG,
|
|
PLAN_GRAMMAR_SPEC,
|
|
check_items_changed,
|
|
milestones_changed,
|
|
missing_check_items,
|
|
new_top_level_completions,
|
|
parse_plan,
|
|
render_plan_grammar_spec,
|
|
restore_missing_check_items,
|
|
subgoal_hash,
|
|
unintended_milestone_edits,
|
|
)
|
|
|
|
PLAN = """# Task Plan
|
|
Some free-form context the parser must ignore.
|
|
- [x] Open the email app
|
|
- [/] [Loop] Inspect candidates (Exit: all inspected; Interval: n/a)
|
|
- [x] Candidate #1
|
|
- [/] Candidate #2
|
|
- [ ] [Loop:continuous] Monitor inbox (Interval: every 5 minutes)
|
|
- [!] Blocked milestone
|
|
"""
|
|
|
|
|
|
def test_parse_plan_machine_channel_only():
|
|
snap = parse_plan(PLAN)
|
|
assert len(snap.items) == 6
|
|
assert len(snap.top_level) == 4
|
|
assert snap.milestone_texts[0] == "Open the email app"
|
|
|
|
top = snap.top_level
|
|
assert top[0].is_done
|
|
assert top[1].is_active and top[1].is_loop and not top[1].is_continuous
|
|
assert top[2].is_pending and top[2].is_loop and top[2].is_continuous
|
|
assert top[3].status == "!"
|
|
|
|
sub = [i for i in snap.items if not i.is_top_level]
|
|
assert all(i.indent == 2 for i in sub)
|
|
|
|
|
|
def test_parse_plan_empty_and_none():
|
|
assert parse_plan(None).items == ()
|
|
assert parse_plan("").items == ()
|
|
assert not parse_plan("just prose, no checkboxes").has_top_level
|
|
|
|
|
|
def test_snapshot_termination_properties():
|
|
assert not parse_plan(PLAN).all_top_level_done
|
|
done = "- [x] A\n- [x] B"
|
|
assert parse_plan(done).all_top_level_done
|
|
assert parse_plan("- [x] A\n- [x] [Loop:continuous] M").continuous_top_level
|
|
|
|
|
|
def test_active_item_navigation():
|
|
snap = parse_plan(PLAN)
|
|
assert snap.first_active().text.startswith("[Loop] Inspect")
|
|
assert snap.last_active().text == "Candidate #2"
|
|
parent = snap.parent_of(snap.last_active())
|
|
assert parent is not None and parent.text.startswith("[Loop] Inspect")
|
|
|
|
|
|
def test_milestones_changed_only_on_text():
|
|
before = parse_plan("- [ ] A\n- [ ] B")
|
|
status_flip = parse_plan("- [x] A\n- [/] B")
|
|
reworded = parse_plan("- [ ] A\n- [ ] B improved")
|
|
added = parse_plan("- [ ] A\n- [ ] B\n- [ ] C")
|
|
assert not milestones_changed(before, status_flip)
|
|
assert milestones_changed(before, reworded)
|
|
assert milestones_changed(before, added)
|
|
|
|
|
|
def test_new_top_level_completions_ordered():
|
|
before = parse_plan("- [x] A\n- [/] B\n- [ ] C")
|
|
after = parse_plan("- [x] A\n- [x] B\n- [x] C")
|
|
assert new_top_level_completions(before, after) == ["B", "C"]
|
|
# Sub-task completions never trigger
|
|
sub_after = parse_plan("- [x] A\n- [/] B\n - [x] B1\n- [ ] C")
|
|
assert new_top_level_completions(before, sub_after) == []
|
|
|
|
|
|
def test_unintended_milestone_edits_pairing():
|
|
before = parse_plan("- [x] A\n- [/] B\n- [ ] C")
|
|
reword_untouched = parse_plan("- [x] A modified\n- [/] B\n- [ ] C")
|
|
assert unintended_milestone_edits(before, reword_untouched) == [("A", "A modified")]
|
|
|
|
# Status changed on the same line: counts as an intentional touch
|
|
complete_and_annotate = parse_plan("- [x] A\n- [x] B (verified)\n- [ ] C")
|
|
assert unintended_milestone_edits(before, complete_and_annotate) == []
|
|
|
|
# Structural change: not this check's business
|
|
removed = parse_plan("- [x] A\n- [/] B")
|
|
assert unintended_milestone_edits(before, removed) == []
|
|
|
|
|
|
def test_subgoal_hash_stability():
|
|
assert subgoal_hash(" A ") == subgoal_hash("A")
|
|
item = parse_plan("- [/] Goal text").items[0]
|
|
assert item.key == subgoal_hash("Goal text")
|
|
|
|
|
|
CHECK_PLAN = """# Task Plan
|
|
- [x] Create the 7:30 AM alarm in the Clock app
|
|
- [ ] open clock app
|
|
- verify: the alarm list shows 7:30 AM with its toggle ON
|
|
- assert: an "Alarm set" toast should appear after creation
|
|
- [ ] Send the report email
|
|
- verify: the email appears in the Sent folder
|
|
- assert@end: no application crash dialog should ever appear
|
|
- verify@end: the device is back on the home screen
|
|
"""
|
|
|
|
|
|
def test_check_lines_parse_with_anchor_and_timing():
|
|
snap = parse_plan(CHECK_PLAN)
|
|
checks = snap.all_check_items
|
|
assert len(checks) == 5
|
|
|
|
alarm = next(i for i in snap.top_level if i.text.startswith("Create the"))
|
|
alarm_checks = snap.check_items_of(alarm)
|
|
assert [c.kind for c in alarm_checks] == ["verify", "assert"]
|
|
assert all(c.when == "on_complete" for c in alarm_checks)
|
|
assert alarm_checks[0].text == "the alarm list shows 7:30 AM with its toggle ON"
|
|
assert all(c.parent_key == alarm.key for c in alarm_checks)
|
|
|
|
task_level = snap.task_level_check_items
|
|
assert len(task_level) == 2
|
|
assert all(c.when == "at_end" for c in task_level)
|
|
assert all(c.parent_key is None for c in task_level)
|
|
assert task_level[0].kind == "assert"
|
|
assert task_level[1].kind == "verify"
|
|
|
|
|
|
def test_check_lines_do_not_disturb_machine_channel():
|
|
"""Check lines must be invisible to milestone hashing, drift detection,
|
|
and completion triggers."""
|
|
without = CHECK_PLAN
|
|
for line in list(without.splitlines()):
|
|
if "verify" in line or ("assert" in line and "- [" not in line):
|
|
without = without.replace(line + "\n", "")
|
|
with_snap = parse_plan(CHECK_PLAN)
|
|
without_snap = parse_plan(without)
|
|
|
|
assert with_snap.milestone_texts == without_snap.milestone_texts
|
|
assert [i.key for i in with_snap.items] == [i.key for i in without_snap.items]
|
|
assert not milestones_changed(without_snap, with_snap)
|
|
assert new_top_level_completions(without_snap, with_snap) == []
|
|
assert unintended_milestone_edits(without_snap, with_snap) == []
|
|
|
|
|
|
def test_check_items_changed_multiset():
|
|
a = parse_plan("- [ ] G\n - verify: X\n - verify: X")
|
|
b = parse_plan("- [ ] G\n - verify: X")
|
|
c = parse_plan("- [ ] G\n - verify: X\n - verify: X")
|
|
assert check_items_changed(a, b)
|
|
assert not check_items_changed(a, c)
|
|
# Status flips never register as check changes
|
|
d = parse_plan("- [x] G\n - verify: X\n - verify: X")
|
|
assert not check_items_changed(a, d)
|
|
|
|
|
|
def test_restore_missing_check_items_reanchors_under_parent():
|
|
before = "- [ ] G1\n - verify: V1\n- [ ] G2\n"
|
|
after = "- [x] G1\n- [ ] G2\n"
|
|
merged = restore_missing_check_items(before, after)
|
|
assert merged is not None
|
|
lines = merged.splitlines()
|
|
assert lines[0] == "- [x] G1"
|
|
assert lines[1] == " - verify: V1"
|
|
assert lines[2] == "- [ ] G2"
|
|
# Idempotent: nothing missing after the merge
|
|
assert restore_missing_check_items(before, merged) is None
|
|
|
|
|
|
def test_restore_missing_check_items_orphan_becomes_task_level_at_end():
|
|
before = "- [ ] G1\n - assert: A1\n- [ ] G2\n"
|
|
after = "- [ ] G2\n" # parent subgoal removed entirely
|
|
merged = restore_missing_check_items(before, after)
|
|
assert merged is not None
|
|
assert merged.splitlines()[-1] == "- assert@end: A1"
|
|
|
|
|
|
def test_restore_missing_check_items_keeps_additions():
|
|
before = "- [ ] G1\n - verify: V1\n"
|
|
after = "- [ ] G1\n - verify: V1\n - verify: V2 (new)\n"
|
|
assert restore_missing_check_items(before, after) is None
|
|
assert not missing_check_items(parse_plan(before), parse_plan(after))
|
|
|
|
|
|
def test_render_plan_grammar_spec_conditional():
|
|
base = render_plan_grammar_spec(include_checks=False)
|
|
extended = render_plan_grammar_spec(include_checks=True)
|
|
# Both-gates-off output carries zero trace of the checking feature
|
|
assert base == PLAN_GRAMMAR_SPEC
|
|
assert "verify" not in base and "assert" not in base
|
|
assert extended.startswith(base)
|
|
assert "- verify:" in extended and "- assert:" in extended and "@end" in extended
|
|
# The capability boundary is declared honestly
|
|
assert "POST-HOC" in extended.upper() or "post-hoc" in extended
|
|
|
|
|
|
def test_grammar_spec_renders_constants():
|
|
assert CONTINUOUS_LOOP_TAG in PLAN_GRAMMAR_SPEC
|
|
assert "[Loop]" in PLAN_GRAMMAR_SPEC
|
|
assert "[/]" in PLAN_GRAMMAR_SPEC and "[x]" in PLAN_GRAMMAR_SPEC
|
|
|
|
|
|
def test_prompts_carry_the_single_sourced_grammar():
|
|
"""Drift guard: the prompts that teach the plan format must splice in the
|
|
machine-enforced grammar spec and reference the formal continuous tag,
|
|
and must no longer key loop protection on Exit wording."""
|
|
agents_dir = Path(__file__).resolve().parents[3] / "artemis" / "agents"
|
|
|
|
planner_raw = (agents_dir / "planner" / "planner.json").read_text(encoding="utf-8")
|
|
operator_raw = (agents_dir / "operator" / "operator.json").read_text(encoding="utf-8")
|
|
|
|
for raw in (planner_raw, operator_raw):
|
|
assert "{{ plan_grammar }}" in raw
|
|
assert "[Loop:continuous]" in raw
|
|
|
|
# The old wording-keyed protection contract must not resurface
|
|
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
|