chenmingxuan/Artemis-Eapil
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 32 Wiki Activity

merge(memory): integrate incremental transcript compaction

Browse Source
This commit is contained in:
somew1nd committed 2026-09-01 17:47:16 -07:00
2 parents f0a4045c49 41ae386a3a
commit 0bcddbf1d7
65 files changed
+9971 -815
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

artemis/agents/operator/operator.json
+1 -1
View File
@@ -1,3 +1,3 @@
{
"main_template": "# ROLE\nYou are the Android Device Operator. Your priority is execution accuracy. You must operate the device step-by-step to achieve the goal, avoiding blind attempts or loops.\n\n# OPERATIONAL PRINCIPLES\n- **Action Scope & Interaction Precision**: Always select gestures whose scope of effect matches the boundaries of the target element. Avoid shortcuts or batch actions from popups or menus if they risk affecting non-target UI components on the screen.[% if \"swipe\" in available_tools %] For any task requiring continuous range interactions, sliding, or boundary selection, prefer using targeted swipe/drag gestures spanning from the element's start bounds [xmin, ymin] to end bounds [xmax, ymax] to ensure the action is strictly confined to the target element. For drag-and-drop, list reordering, or sliding sliders/SeekBars, set duration >= 1000 (e.g. 1500) and always drag slightly PAST the target position to overcome touch slop and reliably trigger the drop/snap/value update. When setting a slider to Maximum (100%) or Minimum (0%), always swipe fully to the extreme boundary to guarantee reaching the absolute limits.[% endif %]\n- **Vision-First Priority**: The screenshot is the primary source of truth for the real device state. Always visually inspect the screen state and layout on the screenshot before referencing the element list. Use the 'Visible UI Elements' list strictly as an auxiliary lookup table to retrieve target element indices and coordinates.\n- **Tool Calling Contract**: Tools that return information needed for the decision ([[ tool_enum(pre_decision_tools, \"and\") ]]) are result-dependent Pre-Decision Tools. Do not submit them in the same tool-call list as a Turn-Ending Action; inspect their result first. The write-through tools `update_note` and `append_note` may accompany at most one Turn-Ending Action when their content is based entirely on evidence already available before that action. Never write an action's expected outcome as if it had already been observed.\n[% if \"wait_for_delay\" in available_tools %]- **Loading & Transitions**: When entering page transitions (e.g., app launching, web page loading, or progress indicators), use `wait_for_delay` to let the screen settle before executing target actions. Some transition states are difficult to discern, such as a thin progress line at the top of the browser screen during webpage loads. Before questioning whether an error has occurred, you can first adopt a waiting strategy to see how the device state evolves. If the state remains ambiguous after waiting, certain UI components might be malfunctioning or failing to load; in this case, consider retrying the action or pivoting to an alternative path.\n[% endif %]- **Memory Preservation & Paginated Exploration**: Using note tools is entirely zero-cost. Actively use this tool to maintain goals and logs. The note tool is the only reliable way to persist memory. Therefore, when you find information on the screen that might be relevant to the task, even if you are not certain, you should first save the information currently visible on the screen before exploring further. If you have already recorded some information, you can use `append_note` to append the new information to the existing content. The note tool can also help you organize lists or information across multiple screens; you can consolidate all information into notes for unified processing.\n- **Large List & Long Text Exploration Strategy (Zero-Miss, Zero-Duplication Single-Pass Principle)**: When traversing large lists, browsing long text bodies, or scanning continuous content feeds, strictly adhere to the single-pass exploration standard:\n 1. **Unidirectional Anchor-Based Traversal (Zero-Miss)**: Maintain a consistent, unidirectional scan (e.g., uniform downward scrolling). Before each scroll, identify the bottom-most visible item/text segment as your visual anchor and carry that anchor plus the scan direction in short-term memory or the traversal ledger. After the scroll, evaluate this handoff before doing anything unrelated: if the anchor remains visible with new content beyond it, record the new content and continue in the same direction without scrolling back to verify; if the anchor is missing, continuity is uncertain, so perform at most one minimal reverse recovery to re-establish overlap, then resume the original direction. While the traversal's declared exit condition remains unresolved, keep this traversal active and do not advance to another milestone or mark it complete.\n 2. **Boundary-Proven Completion**: A single scroll that reveals no new content is not sufficient evidence that the list is complete. For an exhaustive traversal, stop only when an explicit end-of-list indicator is visible, or when one additional successful swipe in the same direction and within the same list container leaves the same final anchor visible with no new items and no loading state. This boundary probe must remain unidLine truncated
"main_template": "# ROLE\nYou are the Android Device Operator. Your priority is execution accuracy. You must operate the device step-by-step to achieve the goal, avoiding blind attempts or loops.\n\n# OPERATIONAL PRINCIPLES\n- **Action Scope & Interaction Precision**: Always select gestures whose scope of effect matches the boundaries of the target element. Avoid shortcuts or batch actions from popups or menus if they risk affecting non-target UI components on the screen.[% if \"swipe\" in available_tools %] For any task requiring continuous range interactions, sliding, or boundary selection, prefer using targeted swipe/drag gestures spanning from the element's start bounds [xmin, ymin] to end bounds [xmax, ymax] to ensure the action is strictly confined to the target element. For drag-and-drop, list reordering, or sliding sliders/SeekBars, set duration >= 1000 (e.g. 1500) and always drag slightly PAST the target position to overcome touch slop and reliably trigger the drop/snap/value update. When setting a slider to Maximum (100%) or Minimum (0%), always swipe fully to the extreme boundary to guarantee reaching the absolute limits.[% endif %]\n- **Vision-First Priority**: The screenshot is the primary source of truth for the real device state. Always visually inspect the screen state and layout on the screenshot before referencing the element list. Use the 'Visible UI Elements' list strictly as an auxiliary lookup table to retrieve target element indices and coordinates.\n- **Tool Calling Contract**: Tools that return information needed for the decision ([[ tool_enum(pre_decision_tools, \"and\") ]]) are result-dependent Pre-Decision Tools. Do not submit them in the same tool-call list as a Turn-Ending Action; inspect their result first. The write-through tools `update_note` and `append_note` may accompany at most one Turn-Ending Action when their content is based entirely on evidence already available before that action. Never write an action's expected outcome as if it had already been observed.\n[% if \"wait_for_delay\" in available_tools %]- **Loading & Transitions**: When entering page transitions (e.g., app launching, web page loading, or progress indicators), use `wait_for_delay` to let the screen settle before executing target actions. Some transition states are difficult to discern, such as a thin progress line at the top of the browser screen during webpage loads. Before questioning whether an error has occurred, you can first adopt a waiting strategy to see how the device state evolves. If the state remains ambiguous after waiting, certain UI components might be malfunctioning or failing to load; in this case, consider retrying the action or pivoting to an alternative path.\n[% endif %]- **Memory Preservation & Paginated Exploration**: Using note tools is entirely zero-cost. Actively use this tool to maintain goals and logs. The note tool is the only reliable way to persist memory. Therefore, when you find information on the screen that might be relevant to the task, even if you are not certain, you should first save the information currently visible on the screen before exploring further. If you have already recorded some information, you can use `append_note` to append the new information to the existing content. The note tool can also help you organize lists or information across multiple screens; you can consolidate all information into notes for unified processing.\n- **Large List & Long Text Exploration Strategy (Zero-Miss, Zero-Duplication Single-Pass Principle)**: When traversing large lists, browsing long text bodies, or scanning continuous content feeds, strictly adhere to the single-pass exploration standard:\n 1. **Unidirectional Anchor-Based Traversal (Zero-Miss)**: Maintain a consistent, unidirectional scan (e.g., uniform downward scrolling). Before each scroll, identify the bottom-most visible item/text segment as your visual anchor and carry that anchor plus the scan direction in short-term memory or the traversal ledger. After the scroll, evaluate this handoff before doing anything unrelated: if the anchor remains visible with new content beyond it, record the new content and continue in the same direction without scrolling back to verify; if the anchor is missing, continuity is uncertain, so perform at most one minimal reverse recovery to re-establish overlap, then resume the original direction. While the traversal's declared exit condition remains unresolved, keep this traversal active and do not advance to another milestone or mark it complete.\n 2. **Boundary-Proven Completion**: A single scroll that reveals no new content is not sufficient evidence that the list is complete. For an exhaustive traversal, stop only when an explicit end-of-list indicator is visible, or when one additional successful swipe in the same direction and within the same list container leaves the same final anchor visible with no new items and no loading state. This boundary probe must remain unidLine truncated
}
artemis/agents/operator/operator.py
+239 -34
View File
@@ -13,7 +13,6 @@
# limitations under the License.
import json
import re
from typing import Any
from uuid import uuid4
@@ -46,12 +45,11 @@ from artemis.utils.coordinates import (
parse_swipe_parameters,
)
from artemis.utils.decorators import wrap_with_callbacks
from artemis.utils.element_hit_test import find_element_at_point
from artemis.utils.logger import get_logger
from artemis.utils.notes import get_note_file_path
from artemis.utils.task_tree import (
build_plan_and_history,
get_active_subgoal_hashes,
)
from artemis.memory.context_policy import build_history_for
from artemis.utils.task_tree import get_active_subgoal_hashes
from artemis.utils.visualization import format_minimal_list_with_elements
logger = get_logger(__name__)
@@ -62,6 +60,7 @@ DEFERRING_TOOLS = {
"read_note",
"list_notes",
"save_note",
"recall_history",
"run_adb_command",
"manage_task",
"analyze_task_output",
@@ -73,14 +72,16 @@ from artemis.agents.operator.prompts import (
PromptComponent,
TemplatePromptComponent,
ObservationPromptComponent,
PlanRecitationPromptComponent,
ScreenshotSimilarityPromptComponent,
HistoricalStateHintPromptComponent,
FeedbackPromptComponent,
CheckItemsExplainerPromptComponent,
BackgroundTasksPromptComponent,
ShortTermMemoryPromptComponent,
TaskPlanWarningPromptComponent,
ToolLimitWarningPromptComponent,
InjectedInstructionPromptComponent,
render_transcript_static_system,
)
from artemis.agents.operator.prompt_builder import load_operator_prompts
@@ -92,6 +93,7 @@ class OperatorNode:
tools: list[BaseTool] | None = None,
prompt_components: list[PromptComponent] | None = None,
last_n_detailed: int = 1,
transcript_config=None,
):
self.ctx = ctx
self.tools = tools or []
@@ -103,6 +105,24 @@ class OperatorNode:
logger.error(f"Failed to load operator prompts: {e}")
self.prompts = {}
# M2 transcript flag (agent.memory.transcript.enabled). Off keeps the
# legacy 2-message prompt path byte-for-byte; an explicit
# ``transcript_config`` overrides the loaded configuration (tests).
if transcript_config is not None:
self._transcript_cfg = transcript_config
else:
try:
from artemis.config import load_agent_config
self._transcript_cfg = load_agent_config().memory.transcript
except Exception:
from artemis.config.agent import MemoryTranscriptConfig
self._transcript_cfg = MemoryTranscriptConfig()
# Index into the rendered message list where this turn's tail begins;
# set by the transcript build and consumed after the tool loop.
self._transcript_turn_base: int | None = None
def _available_device_actions(self) -> frozenset[str]:
"""Device actions the installed actuator backend provides.
@@ -170,6 +190,26 @@ class OperatorNode:
active_background_tasks: list[dict] = None,
newly_finished_tasks: list[dict] = None,
) -> list:
self._transcript_turn_base = None
if getattr(self._transcript_cfg, "enabled", False):
try:
return await self._build_prompt_transcript(
state,
latest_screenshot_b64,
minimal_list,
current_step_num,
steps,
task_plan,
active_background_tasks=active_background_tasks,
newly_finished_tasks=newly_finished_tasks,
)
except Exception as e:
logger.error(
f"Transcript prompt path failed; falling back to the legacy"
f" 2-message build for this turn: {e}"
)
self._transcript_turn_base = None
plan_and_history = self._build_plan_and_history(steps, task_plan)
builder = PromptBuilder()
@@ -187,8 +227,8 @@ class OperatorNode:
(CheckItemsExplainerPromptComponent(), {}),
(ObservationPromptComponent(), {}),
(ScreenshotSimilarityPromptComponent(), {}),
(HistoricalStateHintPromptComponent(), {}),
(InjectedInstructionPromptComponent(), {}),
(ShortTermMemoryPromptComponent(), {}),
(BackgroundTasksPromptComponent(), {}),
(FeedbackPromptComponent(), {}),
(TaskPlanWarningPromptComponent(), {}),
@@ -217,6 +257,174 @@ class OperatorNode:
return builder.build()
# ------------------------------------------------------------------
# Transcript prompt path (M2, flag agent.memory.transcript.enabled)
# ------------------------------------------------------------------
def _ensure_transcript_ledger(self, state=None):
"""Session ledger on the composition root, created on first use."""
from artemis.memory import TranscriptLedger, ensure_step_memory
ledger = getattr(self.ctx, "transcript_ledger", None)
if isinstance(ledger, TranscriptLedger):
return ledger
cfg = self._transcript_cfg
ledger = TranscriptLedger(
step_memory=ensure_step_memory(self.ctx),
image_scrub_depth=getattr(cfg, "image_scrub_depth", 3),
pending_grace_steps=getattr(cfg, "pending_grace_steps", 3),
xml_scrub_depth=getattr(cfg, "xml_scrub_depth", 1),
)
# L2/L3 chunk compression (M3) rides the same flag; without a
# DataEngine there are no step records to chunk, so the ledger simply
# runs scrub-edge-only.
if self.ctx.data_engine is not None:
try:
from artemis.memory import HistoryChunkManager
chunking_cfg = None
try:
from artemis.config import load_agent_config
chunking_cfg = load_agent_config().memory.chunking
except Exception:
pass
ledger.attach_chunker(
HistoryChunkManager(
engine=self.ctx.data_engine,
ctx=self.ctx,
chunking_config=chunking_cfg,
transcript_config=cfg,
goal=getattr(state, "initial_goal", None) if state else None,
)
)
except Exception as e:
logger.error(f"History chunk manager unavailable: {e}")
try:
self.ctx.transcript_ledger = ledger
except Exception:
pass
return ledger
async def _build_prompt_transcript(
self,
state: State,
latest_screenshot_b64: str,
minimal_list: str,
current_step_num: int,
steps: list[dict],
task_plan: str,
active_background_tasks: list[dict] = None,
newly_finished_tasks: list[dict] = None,
) -> list:
"""Build ``S + F + A + tail`` from the session transcript ledger.
The static system message renders once per session (byte-stable S
region); the previous turn is committed into the append-only active
region together with its step key and validator result; the fresh tail
carries the current observation plus the task-plan recitation.
"""
from langchain_core.messages import HumanMessage
ledger = self._ensure_transcript_ledger(state)
# 1. S region: rendered exactly once per session.
if not ledger.has_static_prefix:
static_text = render_transcript_static_system(self.prompts, self.ctx, state)
ledger.set_static_prefix([SystemMessage(content=static_text)])
# 2. F region cold start: an empty ledger over an existing step record
# trail means the process restarted — freeze the compiled history once.
# A staged (not yet committed) turn means this is turn 2+ of a live
# session, not a cold start — step records already exist then, and
# seeding would be rejected by the empty-ledger invariant.
if (
ledger.turn_count == 0
and not ledger.has_staged_turn
and not ledger.has_restored_history
and steps
):
restored = build_history_for(
"operator_cold_start",
task_plan,
steps,
self._get_active_subgoal_hash(),
)
ledger.set_restored_history(
"[Restored history] Rebuilt from the step records of this task"
" after a process restart; step times below are relative to the"
" original session start.\n"
f"{restored}"
)
# 3. Commit the previous turn (its step id and validator result only
# exist now). A turn without a terminal action never ran the validator,
# and ``structured_decisions`` is cleared on planner rejection — the
# gate keeps stale reports out of the ledger.
validator_result = (
getattr(state, "last_execution_result", None)
if getattr(state, "structured_decisions", None)
else None
)
ledger.commit_staged(
step_key=getattr(state, "current_step_id", None),
validator_result=validator_result,
)
# 4. Current tail: observation + plan recitation + injected components.
builder = PromptBuilder()
builder.add_human_content(f"# CURRENT OBSERVATION [{ledger.elapsed_label()}]")
components = [
(PlanRecitationPromptComponent(), {}),
(CheckItemsExplainerPromptComponent(), {}),
(ObservationPromptComponent(), {}),
(ScreenshotSimilarityPromptComponent(), {}),
(HistoricalStateHintPromptComponent(), {}),
(InjectedInstructionPromptComponent(), {}),
(BackgroundTasksPromptComponent(), {}),
(FeedbackPromptComponent(), {}),
(TaskPlanWarningPromptComponent(), {}),
(ToolLimitWarningPromptComponent(), {}),
]
for component, extra_kwargs in components:
kwargs_to_pass = {
"prompts": self.prompts,
"task_plan": task_plan,
"latest_screenshot_b64": latest_screenshot_b64,
"minimal_list": minimal_list,
"current_step_num": current_step_num,
"active_background_tasks": active_background_tasks or [],
"newly_finished_tasks": newly_finished_tasks or [],
"steps": steps,
}
kwargs_to_pass.update(extra_kwargs)
await component(builder, state, self.ctx, **kwargs_to_pass)
tail_content = []
for part in builder.human_parts:
if isinstance(part, str):
tail_content.append({"type": "text", "text": part})
else:
tail_content.append(part)
tail = HumanMessage(content=tail_content)
messages = ledger.render([tail])
self._transcript_turn_base = len(messages) - 1
return messages
def _stage_transcript_turn(self, messages: list) -> None:
"""Hand this turn's tail + tool-loop products to the ledger."""
base = self._transcript_turn_base
self._transcript_turn_base = None
if base is None:
return
try:
ledger = getattr(self.ctx, "transcript_ledger", None)
if ledger is not None:
ledger.stage_turn(messages[base:])
except Exception as e:
logger.error(f"Failed to stage transcript turn: {e}")
async def _invoke_llm_loop(
self,
base_llm,
@@ -729,30 +937,18 @@ class OperatorNode:
state=state,
)
# 8b. Transcript path: hold this turn's messages (observation tail +
# tool-loop products) for commit at the next build, when the turn's
# step id and validator result exist.
self._stage_transcript_turn(messages)
# 9. Update State
structured_decisions = json.dumps(action_result) if action_result else None
short_term_memory = None
combined_thinking = ""
if raw_thinking:
combined_thinking += raw_thinking + "\n"
if native_thinking:
combined_thinking += native_thinking
if combined_thinking:
stm_match = re.search(
r"<short_term_memory>(.*?)</short_term_memory>",
combined_thinking,
re.DOTALL,
)
if stm_match:
short_term_memory = stm_match.group(1).strip()
return {
"structured_decisions": structured_decisions,
"operator_raw_thinking": raw_thinking,
"operator_native_thinking": native_thinking,
"short_term_memory": short_term_memory,
"indexed_points": state.indexed_points,
"indexed_elements": state.indexed_elements,
"current_step_id": state.current_step_id,
@@ -785,14 +981,12 @@ class OperatorNode:
"""Builds a plan checklist and separate execution history."""
active_subgoal_hash = self._get_active_subgoal_hash()
return build_plan_and_history(
return build_history_for(
"operator",
task_plan,
steps,
active_subgoal_hash,
last_n_detailed=self.last_n_detailed,
strict_milestone_pruning=True,
recent_window_size=3,
chronological_last_step=True,
)
def _translate_and_validate_tool(self, tc: dict, state: State) -> tuple[list[dict], str | None]:
@@ -838,7 +1032,7 @@ class OperatorNode:
return None, f"Error: Invalid target index {target}."
indexed_elements = getattr(state, "indexed_elements", None) or []
if 1 <= target_int <= len(indexed_elements):
return indexed_elements[target_int - 1], None
return {**indexed_elements[target_int - 1], "label_source": "index"}, None
return (
None,
(
@@ -851,13 +1045,21 @@ class OperatorNode:
nx, ny = map(float, target)
x = int(max(0, min(width - 1, nx * width / 1000)))
y = int(max(0, min(height - 1, ny * height / 1000)))
# Record-time enrichment: a bare-coordinate target carries no
# element semantics, so hit test the pre-action frame's indexed
# elements to best-effort recover what sits under the point.
hit_el, hit_source = find_element_at_point(
getattr(state, "indexed_elements", None), x, y
)
hit_el = hit_el or {}
return {
"center": [x, y],
"text": None,
"bounds": None,
"class": None,
"resource_id": None,
"is_ocr": False,
"text": hit_el.get("text"),
"bounds": hit_el.get("bounds"),
"class": hit_el.get("class"),
"resource_id": hit_el.get("resource_id"),
"is_ocr": bool(hit_el.get("is_ocr")),
"label_source": hit_source,
}, None
except (ValueError, TypeError):
return (
@@ -894,6 +1096,7 @@ class OperatorNode:
"target_bounds": el.get("bounds"),
"target_resource_id": el.get("resource_id"),
"target_class": el.get("class"),
"target_label_source": el.get("label_source", "none"),
}
], None
@@ -917,6 +1120,7 @@ class OperatorNode:
"target_bounds": el.get("bounds"),
"target_resource_id": el.get("resource_id"),
"target_class": el.get("class"),
"target_label_source": el.get("label_source", "none"),
}
], None
@@ -953,6 +1157,7 @@ class OperatorNode:
"target_bounds": el.get("bounds"),
"target_resource_id": el.get("resource_id"),
"target_class": el.get("class"),
"target_label_source": el.get("label_source", "none"),
}
)
return actions, None
artemis/agents/operator/prompts.py
+224 -39
View File
@@ -52,7 +52,7 @@ from artemis.mcp.action_specs import OPERATOR_SHELL_ORDER
_PRE_DECISION_HELPER_TOOLS = ("ask_explorer", "ask_diagnoser", "video_analyzer")
_PRE_DECISION_ADB_TOOLS = ("run_adb_command", "manage_task", "analyze_task_output")
_PRE_DECISION_MEMORY_TOOLS = ("read_note", "list_notes", "save_note")
_PRE_DECISION_MEMORY_TOOLS = ("read_note", "list_notes", "save_note", "recall_history")
_PRE_DECISION_ALL_TOOLS = (
_PRE_DECISION_HELPER_TOOLS + _PRE_DECISION_ADB_TOOLS + _PRE_DECISION_MEMORY_TOOLS
)
@@ -159,6 +159,119 @@ class PromptComponent:
raise NotImplementedError
def resolve_operator_prompt_tools(ctx: ArtemisContext) -> frozenset[str]:
"""The tool set the operator prompt may advertise for this run.
Extracted verbatim from the template component so the transcript static
prefix (M2) and the legacy per-turn render assemble against the identical
availability set.
"""
# video_analyzer is bound conditionally (graph.py gates it on
# video_recording_tools_enabled); the prompt must not advertise it when the
# tool is not actually available this run.
available = set(OPERATOR_PROMPT_TOOLSET)
setup = getattr(ctx, "execution_setup", None)
if not (setup and getattr(setup, "video_recording_tools_enabled", False)):
available.discard("video_analyzer")
# recall_history (M4) needs a DataEngine to search and is config-gated;
# the prompt must not advertise it when the tool is not bound this run.
if getattr(ctx, "data_engine", None) is None or not _recall_enabled():
available.discard("recall_history")
# Device actions the installed actuator backend does not implement disappear
# from the prompt in lockstep with their tool declarations.
actuator = getattr(ctx, "actuator", None)
if actuator is not None and callable(getattr(actuator, "capabilities", None)):
try:
from artemis.mcp.action_manifest import (
DEVICE_ACTIONS,
available_device_actions,
)
available -= DEVICE_ACTIONS - available_device_actions(actuator)
except Exception as e:
logger.warning(f"Failed to assemble prompt against actuator: {e}")
return frozenset(available)
def _operator_grammar_flags(ctx: ArtemisContext) -> tuple[bool, bool]:
"""(include_checks, verification_active) for the template render."""
setup = getattr(ctx, "execution_setup", None)
# Grammar spec assembly is a function of configuration: with both check
# gates disabled, the check-line grammar never enters any prompt.
include_checks = bool(setup and getattr(setup, "checks_enabled", False))
# The rejection/finding diagnosis trigger only exists while a mechanism
# that can produce rejections or findings is active.
verification_active = include_checks or bool(
setup and not getattr(setup, "disable_planner_validation", True)
)
return include_checks, verification_active
# --- M2 template split -----------------------------------------------------------
# The only volatile span of operator.json's main_template is the plan+history
# section below. The transcript path replaces it with a static pointer so the
# whole system message becomes a byte-stable S region; the legacy path renders
# the untouched template and stays byte-identical.
#: The volatile section of ``main_template`` (must occur exactly once).
PLAN_HISTORY_TEMPLATE_SECTION = "## Current Plan & Execution History\n{{ plan_and_history }}"
#: Static replacement used by the transcript S region.
PLAN_HISTORY_STATIC_POINTER = (
"## Current Plan & Execution History\n"
"Provided in the conversation that follows: earlier turns carry the raw"
" step-by-step execution history (a restored-history block precedes them"
" after a process restart), and each turn's observation message recites"
" the current task plan."
)
def render_transcript_static_system(
prompts: dict,
ctx: ArtemisContext,
state: State,
template_name: str = "main_template",
) -> str:
"""Render the transcript path's byte-stable static system prompt (S region).
Same template, same availability assembly, and the same render inputs as
the legacy path — except the volatile plan+history section is swapped for
a static pointer, so the output never changes across the session.
"""
prompt_template = prompts.get(template_name)
if not prompt_template:
raise KeyError(
"Failed to format prompt, template not found in operator prompts config."
)
if prompt_template.count(PLAN_HISTORY_TEMPLATE_SECTION) != 1:
# Hard dependency of the M2 split (redesign §9): without a clean
# section boundary the S region cannot be byte-stable.
raise ValueError(
"operator main_template no longer contains exactly one plan+history"
" section; the transcript static split cannot be applied."
)
static_template = prompt_template.replace(
PLAN_HISTORY_TEMPLATE_SECTION, PLAN_HISTORY_STATIC_POINTER
)
available = resolve_operator_prompt_tools(ctx)
static_template = apply_operator_prompt_contract(
static_template, available_tools=available
)
include_checks, verification_active = _operator_grammar_flags(ctx)
return Template(static_template).render(
initial_goal=state.initial_goal,
subgoals_status="",
plan_and_history="",
unified_history="",
plan_grammar=render_plan_grammar_spec(include_checks),
verification_active=verification_active,
)
class TemplatePromptComponent(PromptComponent):
async def __call__(self, builder: PromptBuilder, state: State, ctx: ArtemisContext, **kwargs):
prompts = kwargs.get("prompts", {})
@@ -169,42 +282,15 @@ class TemplatePromptComponent(PromptComponent):
"Failed to format prompt, template not found in operator prompts config."
)
# video_analyzer is bound conditionally (graph.py gates it on
# video_recording_tools_enabled); the prompt must not advertise it when the
# tool is not actually available this run.
available = set(OPERATOR_PROMPT_TOOLSET)
setup = getattr(ctx, "execution_setup", None)
if not (setup and getattr(setup, "video_recording_tools_enabled", False)):
available.discard("video_analyzer")
# Device actions the installed actuator backend does not implement disappear
# from the prompt in lockstep with their tool declarations.
actuator = getattr(ctx, "actuator", None)
if actuator is not None and callable(getattr(actuator, "capabilities", None)):
try:
from artemis.mcp.action_manifest import (
DEVICE_ACTIONS,
available_device_actions,
)
available -= DEVICE_ACTIONS - available_device_actions(actuator)
except Exception as e:
logger.warning(f"Failed to assemble prompt against actuator: {e}")
available = resolve_operator_prompt_tools(ctx)
prompt_template = apply_operator_prompt_contract(
prompt_template, available_tools=frozenset(available)
prompt_template, available_tools=available
)
plan_and_history = kwargs.get("plan_and_history", "No plan or history yet.")
# Grammar spec assembly is a function of configuration: with both check
# gates disabled, the check-line grammar never enters any prompt.
include_checks = bool(setup and getattr(setup, "checks_enabled", False))
# The rejection/finding diagnosis trigger only exists while a mechanism
# that can produce rejections or findings is active.
verification_active = include_checks or bool(
setup and not getattr(setup, "disable_planner_validation", True)
)
include_checks, verification_active = _operator_grammar_flags(ctx)
full_prompt = Template(prompt_template).render(
initial_goal=state.initial_goal,
@@ -237,6 +323,21 @@ class ObservationPromptComponent(PromptComponent):
builder.add_human_content(f"--- Visible UI Elements ---\n{minimal_list}")
class PlanRecitationPromptComponent(PromptComponent):
"""Per-turn task-plan recitation for the transcript tail (M2).
With the plan+history section moved out of the (now static) system
message, every observation recites the live task plan; the scrub edge
strips the copies from older turns at depth 1.
"""
async def __call__(self, builder: PromptBuilder, state: State, ctx: ArtemisContext, **kwargs):
from artemis.memory.transcript import PLAN_RECITATION_MARKER
task_plan = kwargs.get("task_plan") or "No task plan yet."
builder.add_human_content(f"{PLAN_RECITATION_MARKER}\n{task_plan}")
class FeedbackPromptComponent(PromptComponent):
"""Append-only injection of source-tagged findings.
@@ -336,14 +437,6 @@ class BackgroundTasksPromptComponent(PromptComponent):
builder.add_human_content("\n".join(lines) + "\n")
class ShortTermMemoryPromptComponent(PromptComponent):
async def __call__(self, builder: PromptBuilder, state: State, ctx: ArtemisContext, **kwargs):
if state.short_term_memory:
builder.add_human_content(
f"--- Short-Term Memory (Scratchpad) ---\n{state.short_term_memory}\n"
)
class TaskPlanWarningPromptComponent(PromptComponent):
async def __call__(self, builder: PromptBuilder, state: State, ctx: ArtemisContext, **kwargs):
steps = kwargs.get("steps", [])
@@ -506,3 +599,95 @@ class ScreenshotSimilarityPromptComponent(PromptComponent):
if diff_count > max_allowed:
break
return diff_count
def _recall_enabled() -> bool:
"""Whether the recall_history tool is enabled by configuration (M4)."""
try:
from artemis.config import load_agent_config
return bool(load_agent_config().memory.recall.enabled)
except Exception:
return True
class HistoricalStateHintPromptComponent(PromptComponent):
"""Local historical-state hint from stored perceptual hashes (M4).
Compares the current screenshot's dHash against the post-action hashes
stamped on every recorded step (``extra_metadata["post_image_dhash"]``,
including steps already chunk-compressed out of the visible transcript) —
an O(n) integer scan, no model call, no historical image bytes. On a
close match to a step *older* than the recent window it injects a one-line
hint pointing at ``recall_history``.
Division of labor with :class:`ScreenshotSimilarityPromptComponent`: that
component's pixel-exact 3-step look-back covers "stuck on the same
screen"; this one covers "returned to a much earlier state". When any of
the last :attr:`RECENT_SILENT_STEPS` steps also matches, this hint stays
silent so the two never fire on the same regime.
"""
#: Matches within this many most-recent steps stay silent (the pixel
#: same-screen note owns that window).
RECENT_SILENT_STEPS: int = 3
#: Upper bound on how many historical steps are scanned (most recent first).
SCAN_CAP: int = 500
async def __call__(self, builder: PromptBuilder, state: State, ctx: ArtemisContext, **kwargs):
latest_screenshot_b64 = kwargs.get("latest_screenshot_b64")
steps = kwargs.get("steps") or []
if not latest_screenshot_b64 or len(steps) <= self.RECENT_SILENT_STEPS:
return
max_distance = 8
try:
from artemis.config import load_agent_config
transcript_cfg = load_agent_config().memory.transcript
if not getattr(transcript_cfg, "similarity_hint", True):
return
max_distance = int(getattr(transcript_cfg, "similarity_max_distance", 8))
except Exception:
pass
try:
from artemis.utils.image_hash import dhash_hex, hamming_distance_hex
if "," in latest_screenshot_b64:
latest_screenshot_b64 = latest_screenshot_b64.split(",", 1)[1]
current_hash = dhash_hex(base64.b64decode(latest_screenshot_b64))
except Exception:
return
if not current_hash:
return
def _step_hash(step: dict) -> str | None:
meta = step.get("extra_metadata") or {}
return meta.get("post_image_dhash") or meta.get("pre_image_dhash")
# Silence rule first: a close match inside the recent window belongs
# to the pixel-level same-screen note, not this hint.
for step in steps[-self.RECENT_SILENT_STEPS :]:
distance = hamming_distance_hex(current_hash, _step_hash(step))
if distance is not None and distance <= max_distance:
return
older_steps = steps[: -self.RECENT_SILENT_STEPS][-self.SCAN_CAP :]
best_step_number = None
best_distance = None
for step in older_steps:
distance = hamming_distance_hex(current_hash, _step_hash(step))
if distance is None or distance > max_distance:
continue
if best_distance is None or distance <= best_distance:
# <= keeps the most recent step on ties.
best_distance = distance
best_step_number = step.get("step_number")
if best_step_number is not None:
builder.add_human_content(
f"Historical state hint: current screen closely resembles the"
f" post-action screen from Step {best_step_number}. Use"
" recall_history only if its details are needed."
)