Artemis-Eapil/tests/unit/utils/test_plan_grammar.py

# 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(midway=False, final=False)
    extended = render_plan_grammar_spec(midway=True, final=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
    # Compatibility shim: ``include_checks`` means "both gates".
    assert render_plan_grammar_spec(include_checks=True) == extended
    assert render_plan_grammar_spec(include_checks=False) == base


def test_render_plan_grammar_spec_words_the_judgment_moment_per_gate():
    """The grammar tells the truth about when a check line is judged: a
    completion-time Checker with a repair loop only exists with midway checks;
    the factory layering (final review only) judges once at task exit."""
    midway = render_plan_grammar_spec(midway=True, final=True)
    assert "at the moment the milestone is marked completed" in midway
    assert "reopens the milestone for repair" in midway

    final_only = render_plan_grammar_spec(midway=False, final=True)
    assert (
        "judged once at task exit from the recorded evidence around the milestone's"
        " completion; there is no midway repair loop" in final_only
    )
    assert "at the moment the milestone is marked completed" not in final_only
    assert "reopens the milestone" not in final_only
    assert "- assert:" in final_only and "@end" in final_only


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(midway=False, final=False)
    midway = render_plan_grammar_spec(midway=True, final=False)
    final_only = render_plan_grammar_spec(midway=False, final=True)
    assert "finding" not in base
    assert "- finding:" in midway and "checker-" in midway
    # Finding lines are projected by the midway checkpoint machinery only.
    assert "- finding:" not in final_only