somew1nd
5d92939d91
Preserve user guidance through history compression, stabilize Flash index targets, and retire withdrawn checks without applying stale verdicts. Improve runtime diagnostics, tool responses, and history rendering. Simplify documentation and comments. Validation: 2076 tests passed in the full suite; 381 affected tests and 26 frontend tests passed after cleanup. Ruff and the quality ratchet passed.
1002 lines
38 KiB
Python
1002 lines
38 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 hashlib
|
|
from artemis.utils.task_tree import (
|
|
format_step_action_result,
|
|
build_plan_and_history,
|
|
get_active_subgoal_hashes,
|
|
get_recent_subgoal_hashes,
|
|
)
|
|
|
|
|
|
def test_build_plan_and_history_separated_layout():
|
|
plan = """- [ ] Open Settings app
|
|
- [ ] Navigate to System settings"""
|
|
|
|
# No steps executed
|
|
output = build_plan_and_history(plan, [], "default")
|
|
assert plan in output
|
|
assert "--- Execution History ---" in output
|
|
assert "No steps executed yet." in output
|
|
|
|
# With steps
|
|
steps = [
|
|
{
|
|
"step_id": "step_1",
|
|
"step_number": 1,
|
|
"relative_time": "2.5s",
|
|
"summary": "Launched settings app",
|
|
"operator_raw_thinking": "Thought 1",
|
|
"action_taken": [{"action": "launch_app"}],
|
|
"last_execution_result": {"status": "success"},
|
|
}
|
|
]
|
|
output = build_plan_and_history(plan, steps, "default", last_n_detailed=1)
|
|
assert plan in output
|
|
assert "--- Execution History ---" in output
|
|
assert "- **Step 1 (Most Recent Step, Start: 2.5s)**" in output
|
|
assert " - Thought 1" in output
|
|
assert " * [Action]: Launched app 'None'" in output
|
|
|
|
|
|
def test_build_plan_and_history_granularities():
|
|
plan = """- [x] Open Settings app
|
|
- [/] Navigate to System settings
|
|
- [ ] Navigate to Languages & input"""
|
|
|
|
subgoal_hash_1 = hashlib.md5(b"Open Settings app").hexdigest()
|
|
subgoal_hash_2 = hashlib.md5(b"Navigate to System settings").hexdigest()
|
|
|
|
steps = [
|
|
{
|
|
"step_id": "step_1",
|
|
"step_number": 1,
|
|
"relative_time": "2.5s",
|
|
"summary": "Launched settings",
|
|
"operator_raw_thinking": "Thought 1",
|
|
"action_taken": [{"action": "launch_app"}],
|
|
"last_execution_result": {"status": "success"},
|
|
"extra_metadata": {"subgoal_hash": subgoal_hash_1},
|
|
},
|
|
{
|
|
"step_id": "step_2",
|
|
"step_number": 2,
|
|
"relative_time": "8.0s",
|
|
"summary": "Swiped down",
|
|
"operator_raw_thinking": "Thought 2",
|
|
"action_taken": [{"action": "swipe"}],
|
|
"last_execution_result": {"status": "success"},
|
|
"extra_metadata": {"subgoal_hash": subgoal_hash_2},
|
|
},
|
|
{
|
|
"step_id": "step_3",
|
|
"step_number": 3,
|
|
"relative_time": "15.0s",
|
|
"summary": "Clicked System settings",
|
|
"operator_raw_thinking": "Thought 3",
|
|
"action_taken": [{"action": "click"}],
|
|
"last_execution_result": {"status": "success"},
|
|
"extra_metadata": {"subgoal_hash": subgoal_hash_2},
|
|
},
|
|
]
|
|
|
|
# Test last_n_detailed = 0 (high-level analysis context, all summarized)
|
|
output_0 = build_plan_and_history(plan, steps, subgoal_hash_2, last_n_detailed=0)
|
|
assert "- *Step 1 (Start: 2.5s): Launched settings*" in output_0
|
|
assert "- *Step 2 (Start: 8.0s): Swiped down*" in output_0
|
|
assert "- *Step 3 (Start: 15.0s): Clicked System settings*" in output_0
|
|
assert "*Thought*: Thought 3" not in output_0
|
|
|
|
# Test last_n_detailed = 1 (normal execution context, only last detailed)
|
|
output_1 = build_plan_and_history(plan, steps, subgoal_hash_2, last_n_detailed=1)
|
|
assert "- *Step 1 (Start: 2.5s): Launched settings*" in output_1
|
|
assert "- *Step 2 (Start: 8.0s): Swiped down*" in output_1
|
|
assert "- **Step 3 (Most Recent Step, Start: 15.0s)**" in output_1
|
|
assert " - Thought 3" in output_1
|
|
|
|
# Test last_n_detailed = 3 (committee context, last 3 detailed)
|
|
output_3 = build_plan_and_history(plan, steps, subgoal_hash_2, last_n_detailed=3)
|
|
assert "- **Step 1 (Start: 2.5s)**" in output_3
|
|
assert " - Thought 1" in output_3
|
|
assert "- **Step 2 (Start: 8.0s)**" in output_3
|
|
assert " - Thought 2" in output_3
|
|
assert "- **Step 3 (Most Recent Step, Start: 15.0s)**" in output_3
|
|
assert " - Thought 3" in output_3
|
|
|
|
# Test keep_subgoal_hashes (milestone filtering for checker/diagnoser)
|
|
# Only keep steps for subgoal_hash_2
|
|
output_filter = build_plan_and_history(
|
|
plan, steps, subgoal_hash_2, keep_subgoal_hashes={subgoal_hash_2}
|
|
)
|
|
assert "Launched settings" not in output_filter # Step 1 is filtered out!
|
|
assert "Swiped down" in output_filter # Step 2 belongs to subgoal_hash_2, kept!
|
|
assert " - Thought 3" in output_filter # Step 3 is the last step, kept detailed!
|
|
|
|
|
|
def test_get_active_subgoal_hashes_fallback():
|
|
# Scenario 1: Normal Active Subgoal (Level 1 [/] active)
|
|
plan_normal = """- [x] Open Settings app
|
|
- [/] Navigate to System settings
|
|
- [ ] Navigate to Languages & input"""
|
|
|
|
h_settings = hashlib.md5(b"Open Settings app").hexdigest()
|
|
h_system = hashlib.md5(b"Navigate to System settings").hexdigest()
|
|
|
|
parent, sub = get_active_subgoal_hashes(plan_normal)
|
|
assert parent == h_system
|
|
assert sub is None
|
|
|
|
# Scenario 2: Indented Sub-subgoal Active (Level 2 [/] active, consolidated to parent)
|
|
plan_indented = """- [x] Open Settings app
|
|
- [/] Navigate to System settings
|
|
- [/] Scroll down to find System settings
|
|
- [ ] Navigate to Languages & input"""
|
|
|
|
parent, sub = get_active_subgoal_hashes(plan_indented)
|
|
assert parent == h_system
|
|
assert sub == hashlib.md5(b"Scroll down to find System settings").hexdigest()
|
|
|
|
# Scenario 3: Safe Fallback (All pending "[ ]", no active subgoal)
|
|
plan_fallback = """- [ ] Open Settings app
|
|
- [ ] Navigate to System settings
|
|
- [ ] Navigate to Languages & input"""
|
|
|
|
parent, sub = get_active_subgoal_hashes(plan_fallback)
|
|
assert parent == h_settings # Successfully fell back to the first subgoal's hash!
|
|
assert sub is None
|
|
|
|
# Scenario 4: All Completed (All "[x]", no active subgoal)
|
|
plan_all_done = """- [x] Open Settings app
|
|
- [x] Navigate to System settings
|
|
- [x] Navigate to Languages & input"""
|
|
|
|
parent, sub = get_active_subgoal_hashes(plan_all_done)
|
|
assert parent == "default"
|
|
assert sub is None
|
|
|
|
|
|
def test_build_plan_and_history_sliding_window():
|
|
plan = """- [x] Subgoal 1
|
|
- [x] Subgoal 2
|
|
- [/] Subgoal 3"""
|
|
|
|
subgoal_hash_1 = hashlib.md5(b"Subgoal 1").hexdigest()
|
|
subgoal_hash_2 = hashlib.md5(b"Subgoal 2").hexdigest()
|
|
subgoal_hash_3 = hashlib.md5(b"Subgoal 3").hexdigest()
|
|
|
|
steps = [
|
|
{
|
|
"step_id": "step_1",
|
|
"step_number": 1,
|
|
"summary": "Step 1 summary",
|
|
"extra_metadata": {"subgoal_hash": subgoal_hash_1},
|
|
},
|
|
{
|
|
"step_id": "step_2",
|
|
"step_number": 2,
|
|
"summary": "Step 2 summary",
|
|
"extra_metadata": {"subgoal_hash": subgoal_hash_1},
|
|
},
|
|
{
|
|
"step_id": "step_3",
|
|
"step_number": 3,
|
|
"summary": "Step 3 summary",
|
|
"extra_metadata": {"subgoal_hash": subgoal_hash_2},
|
|
},
|
|
{
|
|
"step_id": "step_4",
|
|
"step_number": 4,
|
|
"summary": "Step 4 summary",
|
|
"extra_metadata": {"subgoal_hash": subgoal_hash_3},
|
|
},
|
|
{
|
|
"step_id": "step_5",
|
|
"step_number": 5,
|
|
"summary": "Step 5 summary",
|
|
"extra_metadata": {"subgoal_hash": subgoal_hash_3},
|
|
},
|
|
{
|
|
"step_id": "step_6",
|
|
"step_number": 6,
|
|
"summary": "Step 6 summary",
|
|
"extra_metadata": {"subgoal_hash": subgoal_hash_3},
|
|
},
|
|
{
|
|
"step_id": "step_7",
|
|
"step_number": 7,
|
|
"summary": "Step 7 summary",
|
|
"extra_metadata": {"subgoal_hash": subgoal_hash_3},
|
|
},
|
|
]
|
|
|
|
# Case 1: min_summaries = 2
|
|
# subgoal 3 has 4 steps (4, 5, 6, 7). This already exceeds min_summaries=2.
|
|
# So older steps (1, 2, 3) from completed subgoals should be compressed/hidden!
|
|
output_c1 = build_plan_and_history(
|
|
plan, steps, subgoal_hash_3, min_summaries=2, last_n_detailed=0
|
|
)
|
|
assert "Step 1 summary" not in output_c1
|
|
assert "Step 2 summary" not in output_c1
|
|
assert "Step 3 summary" not in output_c1
|
|
assert "Step 4 summary" in output_c1
|
|
assert "Step 5 summary" in output_c1
|
|
assert "Step 6 summary" in output_c1
|
|
assert "Step 7 summary" in output_c1
|
|
|
|
# Case 2: min_summaries = 5
|
|
# subgoal 3 has 4 steps (4, 5, 6, 7), which translates to 3 non-last steps.
|
|
# To reach min_summaries=5 non-last steps, the sliding window slides back to keep Step 3 and Step 2 visible!
|
|
# But Step 1 should still be compressed/hidden.
|
|
output_c2 = build_plan_and_history(
|
|
plan, steps, subgoal_hash_3, min_summaries=5, last_n_detailed=0
|
|
)
|
|
assert "Step 1 summary" not in output_c2
|
|
assert "Step 2 summary" in output_c2 # Kept visible by sliding window!
|
|
assert "Step 3 summary" in output_c2 # Kept visible by sliding window!
|
|
assert "Step 4 summary" in output_c2
|
|
assert "Step 5 summary" in output_c2
|
|
assert "Step 6 summary" in output_c2
|
|
assert "Step 7 summary" in output_c2
|
|
|
|
|
|
def test_build_plan_and_history_clean_success_result():
|
|
plan = "- [ ] Subgoal 1"
|
|
steps = [
|
|
{
|
|
"step_id": "step_1",
|
|
"step_number": 1,
|
|
"relative_time": "2.5s",
|
|
"summary": "Swiped screen",
|
|
"action_taken": [{"action": "swipe"}],
|
|
"last_execution_result": {
|
|
"executed_actions": [{"action": "swipe"}],
|
|
"status": "success",
|
|
},
|
|
}
|
|
]
|
|
output = build_plan_and_history(plan, steps, "default", last_n_detailed=1)
|
|
|
|
# The output should contain "status": "success" but NOT the redundant "executed_actions"
|
|
# Success results are stripped under Factual Realism
|
|
assert "* [Validator Execution Result]" not in output
|
|
assert "executed_actions" not in output
|
|
|
|
|
|
def test_build_plan_and_history_with_tool_calls():
|
|
plan = "- [ ] Subgoal 1"
|
|
steps = [
|
|
{
|
|
"step_id": "step_1",
|
|
"step_number": 1,
|
|
"relative_time": "2.5s",
|
|
"summary": "Read note and swiped",
|
|
"action_taken": [{"action": "swipe"}],
|
|
"last_execution_result": {"status": "success"},
|
|
"tool_calls": [
|
|
{
|
|
"name": "read_note",
|
|
"payload": {
|
|
"args": {"key": "other_note"},
|
|
"result": "- [ ] Milestone",
|
|
},
|
|
"status": "success",
|
|
}
|
|
],
|
|
}
|
|
]
|
|
output = build_plan_and_history(plan, steps, "default", last_n_detailed=1)
|
|
|
|
assert "* [Reasoning & tool calls]:" in output
|
|
assert '`read_note({"key": "other_note"})` -> - [ ] Milestone' in output
|
|
|
|
|
|
def test_build_plan_and_history_with_interleaved_events():
|
|
plan = "- [ ] Subgoal 1"
|
|
steps = [
|
|
{
|
|
"step_id": "step_1",
|
|
"step_number": 1,
|
|
"relative_time": "2.5s",
|
|
"summary": "Read note and thought",
|
|
"action_taken": [{"action": "swipe"}],
|
|
"last_execution_result": {"status": "success"},
|
|
"interleaved_events": [
|
|
{"type": "thought", "content": "I want to search for files."},
|
|
{
|
|
"type": "tool_call",
|
|
"name": "list_notes",
|
|
"args": {"dir": "/notes"},
|
|
"result": ["notes.txt"],
|
|
},
|
|
{"type": "thought", "content": "I see notes.txt, let's read it."},
|
|
{
|
|
"type": "tool_call",
|
|
"name": "read_note",
|
|
"args": {"file": "notes.txt"},
|
|
"result": "Milestone plan",
|
|
},
|
|
],
|
|
}
|
|
]
|
|
output = build_plan_and_history(plan, steps, "default", last_n_detailed=1)
|
|
|
|
# It should render interleaved thoughts and tool calls in natural order
|
|
assert " - I want to search for files." in output
|
|
assert ' - [Tool Call]: `list_notes({"dir": "/notes"})` -> [\'notes.txt\']' in output
|
|
assert " - I see notes.txt, let's read it." in output
|
|
assert ' - [Tool Call]: `read_note({"file": "notes.txt"})` -> Milestone plan' in output
|
|
|
|
# It should use the new section header
|
|
assert "* [Reasoning & tool calls]:" in output
|
|
|
|
|
|
def test_build_plan_and_history_concatenates_thoughts():
|
|
plan = "- [ ] Subgoal 1"
|
|
steps = [
|
|
{
|
|
"step_id": "step_1",
|
|
"step_number": 1,
|
|
"relative_time": "2.5s",
|
|
"summary": "Step with multiple thoughts",
|
|
"action_taken": [{"action": "click", "thought": "Action thought"}],
|
|
"operator_raw_thinking": "Raw thought",
|
|
"last_execution_result": {"status": "success"},
|
|
"interleaved_events": [
|
|
{"type": "thought", "content": "Event thought"},
|
|
],
|
|
}
|
|
]
|
|
output = build_plan_and_history(plan, steps, "default", last_n_detailed=1)
|
|
|
|
# Check that all thoughts are concatenated and visible in the output
|
|
assert " - Action thought" in output
|
|
assert " - Event thought" in output
|
|
assert " - Raw thought" in output
|
|
|
|
|
|
def test_build_plan_and_history_interleaved_decision_loop():
|
|
plan = "- [ ] Subgoal 1"
|
|
steps = [
|
|
{
|
|
"step_id": "step_1",
|
|
"step_number": 1,
|
|
"relative_time": "2.5s",
|
|
"summary": "Completed search",
|
|
"action_taken": [{"action": "click", "target_text": "Search"}],
|
|
"last_execution_result": {"status": "success"},
|
|
"interleaved_events": [
|
|
{"type": "thought", "content": "I should find search icon."},
|
|
{
|
|
"type": "tool_call",
|
|
"name": "ask_explorer",
|
|
"args": {"query": "search icon"},
|
|
"result": {"center": [500, 600]},
|
|
},
|
|
{
|
|
"type": "native_thought",
|
|
"content": "Clicking coordinates [500, 600] now.",
|
|
},
|
|
],
|
|
}
|
|
]
|
|
output = build_plan_and_history(plan, steps, "default", last_n_detailed=1)
|
|
|
|
assert plan in output
|
|
assert "--- Execution History ---" in output
|
|
assert "- **Step 1 (Most Recent Step, Start: 2.5s)**" in output
|
|
assert "* [Reasoning & tool calls]:" in output
|
|
assert " - I should find search icon." in output
|
|
assert (
|
|
'- [Tool Call]: `ask_explorer({"query": "search icon"})` -> {"center":'
|
|
" [500, 600]}" in output
|
|
)
|
|
assert " - Clicking coordinates [500, 600] now." in output
|
|
assert "* [Action]: Tapped 'Search' at None" in output
|
|
assert "* [Validator Execution Result]" not in output
|
|
|
|
|
|
def test_build_plan_and_history_safety_net_and_incident():
|
|
plan = "- [ ] Subgoal 1"
|
|
steps = [
|
|
{
|
|
"step_id": "step_1",
|
|
"step_number": 1,
|
|
"relative_time": "5.0s",
|
|
"summary": "Tried to click search",
|
|
"action_taken": [{"action": "click", "target_text": "Search"}],
|
|
"last_execution_result": {
|
|
"status": "failed",
|
|
"burst": False,
|
|
"execution": [
|
|
{
|
|
"action": "click",
|
|
"target_text": "Search",
|
|
"attempts": [
|
|
"Pre-execution validation failed: Target button is not visible"
|
|
],
|
|
}
|
|
],
|
|
"incident": {
|
|
"kind": "safety_net",
|
|
"category": "target_disappeared",
|
|
"reason": "Target button is not visible",
|
|
"action": {"action": "click", "target_text": "Search"},
|
|
"action_description": "Tapped 'Search' at None",
|
|
"action_index": 0,
|
|
"burst_size": 1,
|
|
"step_number": 1,
|
|
"consecutive_failures": 2,
|
|
"evidence": {},
|
|
},
|
|
},
|
|
"interleaved_events": [
|
|
{"type": "thought", "content": "Let's click search."},
|
|
# Safety net failure
|
|
{
|
|
"type": "tool_call",
|
|
"name": "safety_net_validation",
|
|
"args": {"target": "Search"},
|
|
"result": [
|
|
False,
|
|
"TARGET_DISAPPEARED",
|
|
"Target button is not visible",
|
|
],
|
|
},
|
|
],
|
|
}
|
|
]
|
|
output = build_plan_and_history(plan, steps, "default", last_n_detailed=1)
|
|
|
|
assert "- **Step 1 (Most Recent Step, Start: 5.0s)**" in output
|
|
assert "* [Reasoning & tool calls]:" in output
|
|
assert " - Let's click search." in output
|
|
assert "* [Action]: Tapped 'Search' at None (Intercepted by Pre-Execution Safety Net)" in output
|
|
assert "* [Pre-Execution Safety Net]:" in output
|
|
assert (
|
|
"- [Safety Net Check]: (Intercepted by Pre-Execution Safety Net: Target button is"
|
|
" not visible)" in output
|
|
)
|
|
# No repair agent: the incident IS the result line, with its escalation count.
|
|
assert "Failure Analyzer" not in output
|
|
assert (
|
|
"* [Result]: Error: Intercepted by Pre-Execution Safety Net"
|
|
" (target_disappeared, consecutive failure #2) on action `tap 'Search' at"
|
|
" None`: Target button is not visible" in output
|
|
)
|
|
|
|
|
|
def test_build_plan_and_history_fast_action_burst():
|
|
plan = "- [ ] Subgoal 1"
|
|
burst = [
|
|
{"action": "click", "coordinates": [500, 900]},
|
|
{"action": "click", "coordinates": [880, 120], "target_text": "Skip"},
|
|
{"action": "press_key", "keycode": "BACK"},
|
|
]
|
|
steps = [
|
|
{
|
|
"step_id": "step_1",
|
|
"step_number": 1,
|
|
"relative_time": "2.5s",
|
|
"summary": "Woke the control bar and tapped Skip",
|
|
"action_taken": burst,
|
|
"last_execution_result": {
|
|
"status": "failed",
|
|
"burst": True,
|
|
"execution": [
|
|
{"action": "click", "coordinates": [500, 900]},
|
|
{
|
|
"action": "click",
|
|
"coordinates": [880, 120],
|
|
"target_text": "Skip",
|
|
"attempts": ["Error: tap rejected"],
|
|
},
|
|
{
|
|
"action": "press_key",
|
|
"keycode": "BACK",
|
|
"attempts": ["Skipped (burst aborted)"],
|
|
},
|
|
],
|
|
"incident": {
|
|
"kind": "exec_error",
|
|
"category": "general",
|
|
"reason": "Error: tap rejected",
|
|
"action": burst[1],
|
|
"action_description": "Tapped 'Skip' at [880, 120]",
|
|
"action_index": 1,
|
|
"burst_size": 3,
|
|
"step_number": 1,
|
|
"consecutive_failures": 1,
|
|
"evidence": {},
|
|
},
|
|
},
|
|
}
|
|
]
|
|
# Detailed view: every member with its outcome.
|
|
output = build_plan_and_history(plan, steps, "default", last_n_detailed=1)
|
|
assert "* [Fast-Action Burst]: 3 actions fired back to back without the safety net" in output
|
|
assert " 1. Tapped element at [500, 900] (dispatched)" in output
|
|
assert " 2. Tapped 'Skip' at [880, 120] (FAILED: Error: tap rejected)" in output
|
|
assert " 3. Pressed key 'BACK' (skipped)" in output
|
|
# The failed member is quoted in intent form: it did not happen.
|
|
assert (
|
|
"* [Result]: Error: Execution failed (general, consecutive failure #1) on burst"
|
|
" action 2/3 `tap 'Skip' at [880, 120]`: Error: tap rejected; the remaining"
|
|
" burst actions were not executed" in output
|
|
)
|
|
|
|
# Compact view: the whole burst on one line.
|
|
compact = format_step_action_result(steps[0])
|
|
assert compact.startswith(
|
|
"Fast-action burst (3 actions, unvetted): Tapped element at [500, 900] ->"
|
|
" Tapped 'Skip' at [880, 120] -> Pressed key 'BACK'"
|
|
)
|
|
assert "(Execution failed: Error: tap rejected)" in compact
|
|
|
|
|
|
def test_get_recent_subgoal_hashes_robust_milestone_exclusion(tmp_path):
|
|
# Scenario:
|
|
# 1. Milestone 1 is completed (hash_completed_1).
|
|
# 2. We had a failed attempt 'hash_failed_old' during Milestone 1.
|
|
# 3. Milestone 2 is completed (hash_completed_2).
|
|
# 4. We are now working on Milestone C (hash_active_c).
|
|
# 5. Milestone C had two previous failed renamed versions: Failed Attempt A and Failed Attempt B.
|
|
# 6. subgoal_hash_chain.json maps: A -> B -> C.
|
|
|
|
base_dir = tmp_path
|
|
notes_dir = base_dir / "notes"
|
|
notes_dir.mkdir(parents=True, exist_ok=True)
|
|
chain_path = notes_dir / "subgoal_hash_chain.json"
|
|
|
|
hash_completed_1 = hashlib.md5(b"Completed Milestone 1").hexdigest()
|
|
hash_completed_2 = hashlib.md5(b"Completed Milestone 2").hexdigest()
|
|
hash_active_c = hashlib.md5(b"Active Milestone C").hexdigest()
|
|
|
|
hash_failed_old = hashlib.md5(b"Failed Attempt Old").hexdigest()
|
|
hash_failed_a = hashlib.md5(b"Failed Attempt A").hexdigest()
|
|
hash_failed_b = hashlib.md5(b"Failed Attempt B").hexdigest()
|
|
|
|
# Write the hash chain representing: A -> B -> C
|
|
import json
|
|
|
|
chain_data = {hash_failed_a: hash_failed_b, hash_failed_b: hash_active_c}
|
|
chain_path.write_text(json.dumps(chain_data), encoding="utf-8")
|
|
|
|
steps = [
|
|
# Steps from Completed Milestone 1
|
|
{"step_id": "s1", "extra_metadata": {"subgoal_hash": hash_completed_1}},
|
|
# Old failed attempt during Milestone 1
|
|
{"step_id": "s2", "extra_metadata": {"subgoal_hash": hash_failed_old}},
|
|
# Steps from Completed Milestone 2 (Most recent completed milestone)
|
|
{"step_id": "s3", "extra_metadata": {"subgoal_hash": hash_completed_2}},
|
|
{"step_id": "s4", "extra_metadata": {"subgoal_hash": hash_completed_2}},
|
|
# Steps from Failed Attempt A (Part of the active task slot's past renames)
|
|
{"step_id": "s5", "extra_metadata": {"subgoal_hash": hash_failed_a}},
|
|
{"step_id": "s6", "extra_metadata": {"subgoal_hash": hash_failed_a}},
|
|
# Steps from Failed Attempt B (Part of the active task slot's past renames)
|
|
{"step_id": "s7", "extra_metadata": {"subgoal_hash": hash_failed_b}},
|
|
{"step_id": "s8", "extra_metadata": {"subgoal_hash": hash_failed_b}},
|
|
# Steps from Active Milestone C
|
|
{"step_id": "s9", "extra_metadata": {"subgoal_hash": hash_active_c}},
|
|
]
|
|
|
|
# Test raw transitive alias resolver
|
|
from artemis.utils.task_tree import get_all_subgoal_aliases
|
|
|
|
aliases = get_all_subgoal_aliases(hash_active_c, base_dir)
|
|
assert aliases == {hash_active_c, hash_failed_b, hash_failed_a}
|
|
|
|
keep_hashes = get_recent_subgoal_hashes(steps, hash_active_c, base_dir)
|
|
|
|
# Should keep:
|
|
# 1. The current active subgoal (hash_active_c)
|
|
assert hash_active_c in keep_hashes
|
|
|
|
# 2. The failed/renamed subgoals during active slot (hash_failed_b, hash_failed_a) - resolved transitively via chain
|
|
assert hash_failed_b in keep_hashes
|
|
assert hash_failed_a in keep_hashes
|
|
|
|
# 3. The single most recent completed subgoal (hash_completed_2) for continuity context
|
|
assert hash_completed_2 in keep_hashes
|
|
|
|
# Should prune:
|
|
# 4. Older completed subgoals (hash_completed_1)
|
|
assert hash_completed_1 not in keep_hashes
|
|
|
|
# 5. Older failed subgoals from previously completed milestones (hash_failed_old)
|
|
assert hash_failed_old not in keep_hashes
|
|
|
|
|
|
# --- Step replay (shared renderer for replay_steps / MCP view_step_details) ----------
|
|
|
|
|
|
def _replay_step(tool_result, **overrides):
|
|
step = {
|
|
"step_id": "step_4",
|
|
"step_number": 4,
|
|
"relative_time": "12.0s",
|
|
"summary": "Confirmed the Wi-Fi toggle state via the explorer.",
|
|
"action_taken": {"action": "click", "target": [0.5, 0.5], "target_text": "Save"},
|
|
"last_execution_result": {"status": "success"},
|
|
"interleaved_events": [
|
|
{"type": "thought", "content": "Need to confirm the toggle before saving."},
|
|
{
|
|
"type": "tool_call",
|
|
"name": "ask_explorer",
|
|
"args": {"question": "Is the Wi-Fi toggle on?"},
|
|
"result": tool_result,
|
|
},
|
|
{"type": "tool_call", "name": "click", "args": {"target": [0.5, 0.5]}, "result": "ok"},
|
|
],
|
|
}
|
|
step.update(overrides)
|
|
return step
|
|
|
|
|
|
def test_render_step_replay_shows_summary_thoughts_and_full_tool_results():
|
|
from artemis.utils.task_tree import render_step_replay
|
|
|
|
long_result = "The Wi-Fi toggle is ON and the SSID row reads 'HomeNet'. " * 60 # ~3.5k chars
|
|
out = render_step_replay(_replay_step(long_result))
|
|
|
|
assert "- **Step 4 (Start: 12.0s)**" in out
|
|
assert "[Screen]: Confirmed the Wi-Fi toggle state via the explorer." in out
|
|
assert "Need to confirm the toggle before saving." in out
|
|
# The tool call is replayed verbatim: name, args and the whole result.
|
|
assert "`ask_explorer(" in out
|
|
assert long_result in out
|
|
# Action tools are the planned action, not a tool-call line.
|
|
assert "[Action]: Tapped 'Save' at [0.5, 0.5]" in out
|
|
assert "`click(" not in out
|
|
|
|
|
|
def test_render_step_replay_clamps_only_pathological_results():
|
|
from artemis.utils.task_tree import REPLAY_RESULT_CHARS, render_step_replay
|
|
|
|
huge = "y" * (REPLAY_RESULT_CHARS + 500)
|
|
out = render_step_replay(_replay_step(huge))
|
|
assert huge not in out
|
|
assert ("y" * REPLAY_RESULT_CHARS) + "..." in out
|
|
|
|
|
|
def test_live_window_keeps_the_tight_tool_result_clamp():
|
|
"""The Operator's live context window still clamps tool results tightly;
|
|
only the replay renderer is loose."""
|
|
from artemis.utils.task_tree import LIVE_RESULT_CHARS
|
|
|
|
long_result = "z" * 3000
|
|
out = build_plan_and_history(
|
|
"- [ ] Check toggle", [_replay_step(long_result)], "default", last_n_detailed=1
|
|
)
|
|
assert long_result not in out
|
|
assert ("z" * LIVE_RESULT_CHARS) + "..." in out
|
|
# No screen-description line in the live detailed view.
|
|
assert "[Screen]:" not in out
|
|
|
|
|
|
def test_render_step_replay_reports_missing_screen_description_status():
|
|
from artemis.utils.task_tree import render_step_replay
|
|
|
|
pending = render_step_replay(
|
|
_replay_step("ok", summary=None, extra_metadata={"summary_status": "pending"})
|
|
)
|
|
assert "[Screen]: (screen description pending)" in pending
|
|
failed = render_step_replay(
|
|
_replay_step("ok", summary="", extra_metadata={"summary_status": "failed"})
|
|
)
|
|
assert "[Screen]: (screen description unavailable)" in failed
|
|
assert "[Screen]" not in render_step_replay(_replay_step("ok", summary=None))
|
|
|
|
|
|
def test_render_step_replay_shows_described_images_in_tool_results():
|
|
from artemis.utils.task_tree import render_step_replay
|
|
|
|
result = [
|
|
{"type": "text", "text": "--- pre-action screenshot of Step 1 ---"},
|
|
{"type": "text", "text": "[screenshot: pre-action of Step 1]", "image_name": "ab" * 32},
|
|
]
|
|
out = render_step_replay(_replay_step(result))
|
|
assert "[screenshot: pre-action of Step 1]" in out
|
|
|
|
|
|
def test_render_step_replay_header_uses_the_session_clock():
|
|
"""The replay header carries the same ``T+mm:ss`` offset the prompts and
|
|
``search_history`` use; the legacy ``Start:`` text only remains when no
|
|
session clock / step timestamp exists."""
|
|
from artemis.utils.task_tree import render_step_replay, replay_time_label
|
|
|
|
step = _replay_step("ok", timestamp=1000.0 + 61.0)
|
|
out = render_step_replay(step, session_start=1000.0)
|
|
assert "- **Step 4 (T+01:01)**" in out
|
|
assert "Start:" not in out
|
|
|
|
# No session start (or a non-numeric one, e.g. a mocked reader): fallback.
|
|
assert "- **Step 4 (Start: 12.0s)**" in render_step_replay(step)
|
|
assert "- **Step 4 (Start: 12.0s)**" in render_step_replay(step, session_start=object())
|
|
# A step without a timestamp cannot be placed on the session clock.
|
|
assert replay_time_label(_replay_step("ok"), 1000.0) == "Start: 12.0s"
|
|
|
|
|
|
def test_render_step_replay_labels_are_role_neutral():
|
|
"""Flash has no Operator, so the block labels name what they hold."""
|
|
from artemis.utils.task_tree import render_step_replay
|
|
|
|
out = render_step_replay(_replay_step("ok"))
|
|
assert " * [Reasoning & tool calls]:" in out
|
|
assert " * [Action]: Tapped 'Save' at [0.5, 0.5]" in out
|
|
assert "Operator Decision Loop" not in out
|
|
assert "Planned Action" not in out
|
|
|
|
|
|
def test_render_step_replay_prints_a_failure_once():
|
|
"""A failed step shows its error on the [Result] line only, not appended
|
|
to the action line as well."""
|
|
from artemis.utils.task_tree import render_step_replay
|
|
|
|
out = render_step_replay(
|
|
_replay_step(
|
|
"ok",
|
|
last_execution_result={
|
|
"status": "failed",
|
|
"execution": [{"action": "click", "attempts": ["Error: tap rejected"]}],
|
|
},
|
|
)
|
|
)
|
|
assert " * [Action]: Tapped 'Save' at [0.5, 0.5]\n" in out
|
|
assert " * [Result]: Error: tap rejected" in out
|
|
assert out.count("tap rejected") == 1
|
|
assert "Execution failed:" not in out
|
|
|
|
|
|
def test_render_step_replay_indents_multiline_reasoning():
|
|
from artemis.utils.task_tree import render_step_replay
|
|
|
|
step = _replay_step(
|
|
"ok",
|
|
interleaved_events=[
|
|
{"type": "thought", "content": "- Save findings first.\nTap Wi-Fi.\nThen verify."}
|
|
],
|
|
)
|
|
out = render_step_replay(step)
|
|
assert "\n - - Save findings first.\n Tap Wi-Fi.\n Then verify." in out
|
|
|
|
|
|
def test_format_action_clean_directional_swipe_shows_direction_first():
|
|
"""``swipe(direction="up")`` replays as the direction the agent issued; the
|
|
recorded path is a parenthesised detail (Flash records both)."""
|
|
from artemis.utils.task_tree import format_action_clean
|
|
|
|
flash_record = {
|
|
"action": "swipe",
|
|
"coordinates": [600, 700, 600, 300],
|
|
"coordinate_space": "normalized",
|
|
"normalized_start_coordinates": [600, 700],
|
|
"normalized_end_coordinates": [600, 300],
|
|
"args": {"direction": "up", "duration": 800},
|
|
}
|
|
assert (
|
|
format_action_clean(flash_record) == "Swiped up (from [600, 700] to [600, 300]) over 800ms"
|
|
)
|
|
assert format_action_clean({"action": "swipe", "direction": "down"}) == "Swiped down"
|
|
assert (
|
|
format_action_clean(
|
|
{
|
|
"action": "swipe",
|
|
"gesture": "left",
|
|
"start_coordinates": [750, 500],
|
|
"end_coordinates": [250, 500],
|
|
}
|
|
)
|
|
== "Swiped left (from [750, 500] to [250, 500])"
|
|
)
|
|
# Without a direction the path is the headline, as before.
|
|
assert (
|
|
format_action_clean({"action": "swipe", "coordinates": [600, 700, 600, 300]})
|
|
== "Swiped from [600, 700] to [600, 300]"
|
|
)
|
|
|
|
|
|
def test_action_intent_phrase_and_incident_line_use_intent_form():
|
|
"""Failure contexts describe what the agent tried to do, not an outcome."""
|
|
from artemis.utils.task_tree import (
|
|
action_intent_phrase,
|
|
format_action_intent,
|
|
format_incident_clean,
|
|
)
|
|
|
|
assert format_action_intent({"action": "launch_app", "app_name": "NonexistentApp"}) == (
|
|
"launch app 'NonexistentApp'"
|
|
)
|
|
assert (
|
|
format_action_intent(
|
|
{"action": "click", "coordinates": [500, 520], "target_description": "Wi-Fi row"}
|
|
)
|
|
== "tap 'Wi-Fi row' (self-described) at [500, 520]"
|
|
)
|
|
assert format_action_intent({"action": "swipe", "direction": "up"}) == "swipe up"
|
|
assert (
|
|
format_action_intent(
|
|
{
|
|
"action": "input_text",
|
|
"text": "hello",
|
|
"target_text": "Search",
|
|
"coordinates": [1, 2],
|
|
}
|
|
)
|
|
== "type 'hello' into 'Search' at [1, 2]"
|
|
)
|
|
assert format_action_intent({"action": "press_key", "keycode": "BACK"}) == "press key 'BACK'"
|
|
assert format_action_intent({"action": "wait_for_delay", "delay_ms": 200}) == "wait 200ms"
|
|
assert format_action_intent({"action": "long_press", "coordinates": [1, 2]}) == (
|
|
"long press element at [1, 2]"
|
|
)
|
|
assert format_action_intent({"action": "click", "coordinates": [1, 2], "times": 2}) == (
|
|
"double tap element at [1, 2]"
|
|
)
|
|
# Unknown phrases pass through untouched.
|
|
assert action_intent_phrase("Fast-action burst (2 actions, unvetted): x") == (
|
|
"Fast-action burst (2 actions, unvetted): x"
|
|
)
|
|
assert action_intent_phrase("") == ""
|
|
|
|
line = format_incident_clean(
|
|
{
|
|
"kind": "exec_error",
|
|
"category": "general",
|
|
"consecutive_failures": 2,
|
|
"action": {"action": "launch_app", "app_name": "NonexistentApp"},
|
|
"action_description": "Launched app 'NonexistentApp'",
|
|
"reason": "Error finding package for app: NonexistentApp",
|
|
}
|
|
)
|
|
assert line == (
|
|
"Error: Execution failed (general, consecutive failure #2) on action"
|
|
" `launch app 'NonexistentApp'`: Error finding package for app: NonexistentApp"
|
|
)
|
|
intercepted = format_incident_clean(
|
|
{
|
|
"kind": "safety_net",
|
|
"category": "target_disappeared",
|
|
"action": {"action": "click", "target_text": "Skip", "coordinates": [880, 120]},
|
|
"reason": "Target element 'Skip' was not found on the screen.",
|
|
}
|
|
)
|
|
assert intercepted.startswith(
|
|
"Error: Intercepted by Pre-Execution Safety Net (target_disappeared, consecutive"
|
|
" failure #1) on action `tap 'Skip' at [880, 120]`:"
|
|
)
|
|
|
|
|
|
def test_format_action_clean_reads_flash_arguments_and_maps_manage_app():
|
|
"""Read Flash action arguments from ``args`` when formatting app launches."""
|
|
from artemis.utils.task_tree import format_action_clean
|
|
|
|
assert (
|
|
format_action_clean(
|
|
{
|
|
"action": "manage_app",
|
|
"coordinates": None,
|
|
"args": {"action": "launch", "app_name": "com.android.settings"},
|
|
}
|
|
)
|
|
== "Launched app 'com.android.settings'"
|
|
)
|
|
assert (
|
|
format_action_clean({"action": "manage_app", "args": {"action": "stop", "app_name": "x"}})
|
|
== "Stopped app 'x'"
|
|
)
|
|
# Pro's own shape keeps rendering as before.
|
|
assert format_action_clean({"action": "launch_app", "app_name": "x"}) == "Launched app 'x'"
|
|
assert (
|
|
format_action_clean({"action": "manage_app", "app_name": "x", "intent": "clear"})
|
|
== "Managed app 'x' (action: clear)"
|
|
)
|
|
# Top-level fields win over args; args only fill the gaps.
|
|
assert (
|
|
format_action_clean(
|
|
{
|
|
"action": "click",
|
|
"coordinates": [1, 2],
|
|
"target_text": "Save",
|
|
"args": {"target": [9, 9]},
|
|
}
|
|
)
|
|
== "Tapped 'Save' at [1, 2]"
|
|
)
|
|
|
|
|
|
def test_format_action_clean_marks_self_described_coordinate_targets():
|
|
"""A coordinate target's label is the model's own description, and every
|
|
later reader sees it marked as such; an index target's observed element
|
|
text carries no marker."""
|
|
from artemis.utils.task_tree import SELF_DESCRIBED_MARKER, format_action_clean
|
|
|
|
assert SELF_DESCRIBED_MARKER == "(self-described)"
|
|
assert (
|
|
format_action_clean(
|
|
{"action": "tap", "coordinates": [540, 1440], "target_description": "play button"}
|
|
)
|
|
== "Tapped 'play button' (self-described) at [540, 1440]"
|
|
)
|
|
# Observed text outranks the description when both are present, unmarked.
|
|
assert (
|
|
format_action_clean(
|
|
{
|
|
"action": "tap",
|
|
"coordinates": [1, 2],
|
|
"target_text": "Play",
|
|
"target_description": "play button",
|
|
}
|
|
)
|
|
== "Tapped 'Play' at [1, 2]"
|
|
)
|
|
assert (
|
|
format_action_clean(
|
|
{
|
|
"action": "long_press_on",
|
|
"coordinates": [1, 2],
|
|
"duration": 1500,
|
|
"target_description": "app icon",
|
|
}
|
|
)
|
|
== "Long pressed 'app icon' (self-described) at [1, 2] for 1500ms"
|
|
)
|
|
assert (
|
|
format_action_clean(
|
|
{
|
|
"action": "focus_and_input_text",
|
|
"coordinates": [3, 4],
|
|
"text": "hello",
|
|
"target_description": "search box",
|
|
}
|
|
)
|
|
== "Inputted 'hello' into 'search box' (self-described) at [3, 4]"
|
|
)
|
|
assert (
|
|
format_action_clean(
|
|
{
|
|
"action": "swipe",
|
|
"coordinates": [100, 500, 900, 500],
|
|
"duration": 800,
|
|
"target_description": "brightness knob",
|
|
}
|
|
)
|
|
== "Swiped 'brightness knob' (self-described) from [100, 500] to [900, 500] over 800ms"
|
|
)
|
|
# Flash keeps the description under ``args``; it is read from there too.
|
|
assert (
|
|
format_action_clean(
|
|
{
|
|
"action": "click",
|
|
"coordinates": [320, 399],
|
|
"args": {"target": [320, 399], "target_description": "skip ad"},
|
|
}
|
|
)
|
|
== "Tapped 'skip ad' (self-described) at [320, 399]"
|
|
)
|
|
assert format_action_clean(
|
|
{
|
|
"action": "click_sequence",
|
|
"coordinates": [[500, 280], [885, 362]],
|
|
"target_descriptions": ["video body", "skip button"],
|
|
}
|
|
) == (
|
|
"Tapped sequence of targets: 'video body' (self-described) at [500, 280],"
|
|
" 'skip button' (self-described) at [885, 362]"
|
|
)
|
|
# Unnamed targets stay unnamed: no marker without a description.
|
|
assert format_action_clean({"action": "tap", "coordinates": [1, 2]}) == (
|
|
"Tapped element at [1, 2]"
|
|
)
|