diff --git a/artemis/agents/planner/planner.py b/artemis/agents/planner/planner.py index f1013c1..61d0be4 100644 --- a/artemis/agents/planner/planner.py +++ b/artemis/agents/planner/planner.py @@ -46,12 +46,42 @@ from artemis.tools.tool_wrapper import ( from artemis.utils.decorators import wrap_with_callbacks from artemis.utils.logger import get_logger from artemis.utils.notes import get_note_file_path -from artemis.utils.plan_grammar import render_plan_grammar_spec +from artemis.utils.plan_grammar import ( + CHECKBOX_LINE_RE, + STATUS_CHARS, + parse_plan, + render_plan_grammar_spec, +) from artemis.memory.context_policy import build_history_for logger = get_logger(__name__) +def validate_plan_format(content: str) -> tuple[bool, str]: + """Validates a freshly written plan against the machine-channel grammar. + + Single-sourced from :mod:`artemis.utils.plan_grammar`: any status character + the grammar accepts (including in-progress ``[/]``) is legal here too, so + the Planner can never reject a line the rest of the harness would parse. + """ + for line in content.splitlines(): + stripped = line.strip() + if stripped.startswith("- [") and not CHECKBOX_LINE_RE.match(stripped): + allowed = ", ".join(f"'[{c}]'" for c in STATUS_CHARS) + return ( + False, + f"Invalid task status in line: {line}. Must be one of {allowed}.", + ) + + if not parse_plan(content).items: + return ( + False, + "Plan must contain at least one subgoal starting with '- [ ]'.", + ) + + return True, "" + + def _checks_enabled(ctx: ArtemisContext) -> bool: """Config-driven prompt assembly gate: with both check gates disabled, no check-related instruction reaches any planner prompt.""" @@ -276,38 +306,6 @@ class PlannerNode: llm = llm.bind_tools(all_tools) llm_fallback = llm_fallback.bind_tools(all_tools) - def validate_plan_format(content: str) -> tuple[bool, str]: - lines = content.splitlines() - has_subgoal = False - for line in lines: - stripped = line.strip() - if stripped.startswith("- ["): - has_subgoal = True - if not ( - stripped.startswith("- [ ]") - or stripped.startswith("- [x]") - or stripped.startswith("- [!]") - ): - return ( - False, - ( - f"Invalid task status in line: {line}. Must be" - " '[ ]', '[x]', or '[!]'." - ), - ) - elif stripped.startswith("#") or not stripped: - continue - else: - continue - - if not has_subgoal: - return ( - False, - ("Plan must contain at least one subgoal starting with '- [ ]'."), - ) - - return True, "" - current_messages = messages max_validation_attempts = 3 max_iterations = 10 diff --git a/artemis/config/agent.py b/artemis/config/agent.py index b017e7f..461e0e6 100644 --- a/artemis/config/agent.py +++ b/artemis/config/agent.py @@ -116,7 +116,7 @@ class PlannerValidationConfig(BaseModel): """Configuration specific to Planner validation of milestone changes.""" enabled: bool = Field( - default=False, + default=True, description="Whether to enable async planner validation when task plan milestones change.", ) similarity_threshold: float = Field( @@ -146,15 +146,18 @@ class CheckerConfig(BaseModel): """Configuration for plan-driven checkpoint verification and the final review.""" enabled: bool = Field( - default=False, + default=True, description=( "Master switch (compat alias): False disables BOTH midway" " checkpoints and the final check." ), ) midway_checks: bool = Field( - default=True, - description="Whether plan-declared midway checkpoints run (requires enabled=True).", + default=False, + description=( + "Whether plan-declared midway checkpoints run (requires enabled=True)." + " Off in the factory layering: final check on, midway checks off." + ), ) final_check: bool = Field( default=True, diff --git a/artemis/context.py b/artemis/context.py index a82d297..68d22af 100644 --- a/artemis/context.py +++ b/artemis/context.py @@ -105,10 +105,13 @@ class ExecutionSetup(BaseModel): enable_remote_tracing: bool = False app_lock_status: AppLaunchResult | None = None video_recording_tools_enabled: bool = Field(default_factory=detect_video_tools_enabled) - disable_checker: bool = True + disable_checker: bool = False """Legacy master switch (compat alias): ``True`` disables BOTH the midway - checkpoints and the final check, regardless of the individual gates below.""" - disable_midway_checks: bool = False + checkpoints and the final check, regardless of the individual gates below. + Factory default is ``False``: the exit final review runs out of the box.""" + disable_midway_checks: bool = True + """Midway checkpoints are OFF by default (factory layering: final check on, + planner validation on, midway checks off).""" disable_final_check: bool = False checker_max_iterations: int = 20 final_check_max_attempts: int = 3 @@ -118,7 +121,7 @@ class ExecutionSetup(BaseModel): settlement_timeout: float = 120.0 assert_failure_policy: Literal["continue", "halt"] = "continue" disable_device_probes: bool = False - disable_planner_validation: bool = True + disable_planner_validation: bool = False planner_validation_threshold: float = 0.85 enable_committee: bool = False committee_debate_rounds: int = 2 diff --git a/artemis/graph/checkpoints.py b/artemis/graph/checkpoints.py index 044584e..1a91857 100644 --- a/artemis/graph/checkpoints.py +++ b/artemis/graph/checkpoints.py @@ -41,11 +41,14 @@ from typing import Any, Literal from pydantic import BaseModel, Field from artemis.utils.logger import get_logger -from artemis.utils.notes import get_note_file_path +from artemis.utils.notes import append_note_content, get_note_file_path from artemis.utils.plan_grammar import ( CHECKBOX_LINE_RE, + STATUS_ACTIVE, + STATUS_DONE, CheckItem, PlanSnapshot, + apply_finding_lines, parse_plan, subgoal_hash, ) @@ -220,8 +223,12 @@ def revert_subgoal_status(ctx, subgoal_key: str) -> bool: if not match or match.group("indent"): continue if subgoal_hash(match.group("text").strip()) == subgoal_key: - if match.group("status") == "x": - lines[idx] = re.sub(r"^(\s*-\s*\[)x(\])", r"\g<1>/\g<2>", line) + if match.group("status") == STATUS_DONE: + lines[idx] = re.sub( + r"^(\s*-\s*\[)" + re.escape(STATUS_DONE) + r"(\])", + lambda m: m.group(1) + STATUS_ACTIVE + m.group(2), + line, + ) changed = True break if changed: @@ -232,6 +239,127 @@ def revert_subgoal_status(ctx, subgoal_key: str) -> bool: return False +# --- Verify-finding persistence (four layers; redesign "verify finding") ------------- +# +# Layer 1: the append-only ledger above (single source of truth — unchanged). +# Layer 2: `- finding:` plan lines — a single-direction projection rendered by +# :func:`sync_finding_lines`; harness code never reads them back. +# Layer 3: the system-authored `checker-` note — full details on +# first failure, then one appended section per attempt verdict. +# Layer 4: single-turn ``operator_feedback`` findings (unchanged). + +#: Reserved note-key prefixes for system-authored checker repair logs. The +#: model-side note tool wrappers reject writes to these keys. ``checker-`` is +#: the prefix actually used for storage (a colon is not a legal filename +#: character on Windows/NTFS); ``checker:`` is reserved alongside it so the +#: conceptual spelling can never be squatted either. +CHECKER_NOTE_PREFIXES = ("checker-", "checker:") + +_FINDING_HEADLINE_EVIDENCE_LIMIT = 160 + + +def checker_note_key(checkpoint_id: str) -> str: + """Note key of the system-authored detail/repair log for one checkpoint.""" + return f"checker-{checkpoint_id}" + + +def is_checker_note_key(key: str) -> bool: + """True when a note key is reserved for system-authored checker logs.""" + return str(key or "").strip().lower().startswith(CHECKER_NOTE_PREFIXES) + + +def _finding_registry(ctx) -> dict: + """checkpoint_id -> rendered headline of the unresolved verify finding. + + Kept as a dynamic context attribute (``ArtemisContext`` allows extras) so + the registry lives exactly as long as the run. + """ + registry = getattr(ctx, "checker_findings", None) + if not isinstance(registry, dict): + registry = {} + try: + ctx.checker_findings = registry + except Exception: + pass + return registry + + +def _render_finding_headline(checkpoint_id: str, verify_failures: list) -> str: + first = verify_failures[0] + evidence = str(getattr(first, "evidence", "") or "").strip() + if len(evidence) > _FINDING_HEADLINE_EVIDENCE_LIMIT: + evidence = evidence[: _FINDING_HEADLINE_EVIDENCE_LIMIT - 1] + "…" + detail = f": {evidence}" if evidence else "" + return ( + f"verify failed — '{first.item_text}'{detail}" + f" (details & repair log: note '{checker_note_key(checkpoint_id)}')" + ) + + +def sync_finding_lines(ctx) -> None: + """Re-projects the unresolved verify findings into ``task_plan.md``. + + Deterministic merge-back protection for the ``- finding:`` channel: every + call strips all finding lines and re-renders one per unresolved checkpoint + under its subgoal, so a model deletion grows back and a resolved finding + disappears. Pure projection — nothing is ever parsed back out of the plan. + """ + if not getattr(ctx, "data_engine", None): + return + path = get_note_file_path(ctx.data_engine.base_dir, "task_plan") + if not path.exists(): + return + try: + content = path.read_text(encoding="utf-8") + registry = getattr(ctx, "checker_findings", None) + findings = dict(registry) if isinstance(registry, dict) else {} + projected = apply_finding_lines(content, findings) + if projected != content: + path.write_text(projected, encoding="utf-8") + except Exception as e: + logger.error(f"Failed to project finding lines into task plan: {e}") + + +def _log_checker_note(ctx, run: CheckpointRun, verdicts: list) -> None: + """Appends this attempt's verdicts to the checkpoint's system note. + + The note starts at the FIRST failed verify (full criterion/evidence/ + suggestion detail) and afterwards accumulates one section per attempt — + including the eventually passing one — as a persistent repair log. It is + written through the notes storage API directly (never the model tool + path), stays on disk after resolution, and is discoverable via read_note / + recall_history under the key returned by :func:`checker_note_key`. + """ + try: + base_dir = ctx.data_engine.base_dir + cid = run.checkpoint.checkpoint_id + key = checker_note_key(cid) + note_path = get_note_file_path(base_dir, key) + verify_failures = [v for v in verdicts if v.kind == "verify" and v.status == "failed"] + if not note_path.exists() and not verify_failures: + return # the log only starts once a verify criterion has failed + + blocks: list[str] = [] + if not note_path.exists(): + blocks.append( + f"# Checker findings — subgoal: {run.checkpoint.subgoal_text}\n" + "System-authored verification log (append-only). Each section" + " records one checkpoint attempt's verdicts." + ) + outcome = "failed" if verify_failures else "passed" + lines = [f"## attempt {run.attempt_id} — verify {outcome}"] + for v in verdicts: + line = f"- [{v.kind}] '{v.item_text}': {v.status} — {v.evidence}" + suggestion = str(getattr(v, "suggestion", "") or "") + if suggestion: + line += f"\n suggestion: {suggestion}" + lines.append(line) + blocks.append("\n".join(lines)) + append_note_content(base_dir, key, "\n\n".join(blocks)) + except Exception as e: + logger.error(f"Failed to append checker note for {run.attempt_id}: {e}") + + # --- Queueing (called from _process_plan_write; never spawns) ------------------------ @@ -390,6 +518,8 @@ def harvest_run( verdicts = list(getattr(report, "verdicts", []) or []) _record_verdicts(ctx, run, verdicts, anchor_step_id) _annotate_history_chunks(ctx, run, verdicts) + # Layer 3: the per-checkpoint repair log accumulates every booked attempt. + _log_checker_note(ctx, run, verdicts) # Applicability gate for side effects: the verdict must belong to the # current attempt of its checkpoint AND the anchored subgoal's text must be @@ -400,12 +530,21 @@ def harvest_run( verify_failures = [v for v in verdicts if v.kind == "verify" and v.status == "failed"] assert_failures = [v for v in verdicts if v.kind == "assert" and v.status == "failed"] + registry = _finding_registry(ctx) + registry_changed = False + if verify_failures and applicable and not user_stopped: max_repairs = int(_setting(ctx, "checkpoint_max_repairs", 2)) repairs = ctx.checkpoint_repairs.get(checkpoint.checkpoint_id, 0) if repairs < max_repairs: ctx.checkpoint_repairs[checkpoint.checkpoint_id] = repairs + 1 reverted = revert_subgoal_status(ctx, checkpoint.checkpoint_id) + # Layer 2: register the standing plan headline for this unresolved + # finding; the projection below renders it under the subgoal. + registry[checkpoint.checkpoint_id] = _render_finding_headline( + checkpoint.checkpoint_id, verify_failures + ) + registry_changed = True for v in verify_failures: suggestion = getattr(v, "suggestion", "") or "" findings.append( @@ -420,11 +559,22 @@ def harvest_run( " it again." ) else: + # Quota exhausted: the run is settled for this checkpoint — the + # standing headline is retired (the ledger and the checker note + # keep the full record). + registry_changed = registry.pop(checkpoint.checkpoint_id, None) is not None logger.warning( f"Checkpoint {checkpoint.checkpoint_id} exhausted its repair" f" quota ({max_repairs}); verdict stays failed, no further" " repair loop." ) + elif not verify_failures: + # A booked attempt without failed verify criteria resolves (or + # releases) the checkpoint: its standing headline is retired. + registry_changed = registry.pop(checkpoint.checkpoint_id, None) is not None + + if registry_changed: + sync_finding_lines(ctx) if assert_failures: policy = str(_setting(ctx, "assert_failure_policy", "continue")) diff --git a/artemis/graph/graph.py b/artemis/graph/graph.py index 3da68c8..252b3d8 100644 --- a/artemis/graph/graph.py +++ b/artemis/graph/graph.py @@ -42,11 +42,13 @@ from artemis.graph.checkpoints import ( final_check_enabled, harvest_finished_checkpoints, has_ledger_records, + is_checker_note_key, queue_checkpoints, read_ledger, revert_subgoal_status, settle_all_checkpoints, spawn_pending_checkpoints, + sync_finding_lines, write_run_outcome, ) from artemis.graph.perception import perception_node @@ -518,6 +520,27 @@ def check_unintended_rewrite(content_before: str, content_after: str) -> str | N ) +def _reject_checker_note_write(key: str, tool_call_id: str | None) -> ToolMessage | None: + """Guard for the system-authored checker note channel. + + Note keys with the reserved ``checker-`` / ``checker:`` prefix are written + exclusively by the checkpoint verifier (repair logs); model-side writes are + rejected with guidance instead of silently colliding with the log. + """ + if not is_checker_note_key(key): + return None + return ToolMessage( + content=( + f"The note key '{key}' was not written: keys with the 'checker-'" + " prefix are reserved for system-authored verification repair" + " logs. Those notes are read-only for you — read them with" + " read_note. Record your own observations under a different key." + ), + tool_call_id=tool_call_id or "", + status="success", + ) + + class NoteArgs(BaseModel): model_config = {"ignored_types": (CyFunctionDetector,)} key: str = Field(..., description=SAVE_NOTE_ARG_KEY_DESC) @@ -570,6 +593,13 @@ async def _process_plan_write( except Exception as e: logger.error(f"Failed to write merged check lines: {e}") + # Deterministic finding-line projection: unresolved verify findings are + # re-rendered from the checkpoint repair state on every plan write, so a + # model deletion grows back and a resolved finding is retired. Finding + # lines live outside the checkbox/check machine channel, so this rewrite + # never influences the parses below. + sync_finding_lines(ctx) + before = parse_plan(content_before) after = parse_plan(content_after) @@ -650,6 +680,9 @@ def wrap_note_tool(ctx: ArtemisContext, original_tool): state: Annotated[State, InjectedState] = None, ): """Wrapped tool to intercept task plan updates.""" + rejection = _reject_checker_note_write(key, tool_call_id) + if rejection is not None: + return rejection args = {"key": key, "content": content} if key != "task_plan" or not ctx.data_engine: return await invoke_tool_with_injection( @@ -706,6 +739,9 @@ def wrap_update_note_tool(ctx: ArtemisContext, original_tool): state: Annotated[State, InjectedState] = None, ): """Wrapped tool to intercept task plan updates via update_note.""" + rejection = _reject_checker_note_write(key, tool_call_id) + if rejection is not None: + return rejection args = {"key": key, "target": target, "replacement": replacement} if key != "task_plan" or not ctx.data_engine: return await invoke_tool_with_injection( diff --git a/artemis/sdk/builders/agent_config_builder.py b/artemis/sdk/builders/agent_config_builder.py index 5228db4..2bcd3f8 100644 --- a/artemis/sdk/builders/agent_config_builder.py +++ b/artemis/sdk/builders/agent_config_builder.py @@ -67,7 +67,7 @@ class AgentConfigBuilder: self._graph_config_callbacks: Callbacks = None self._video_recording_tools_enabled: bool = detect_video_tools_enabled() self._force_web_accessibility: bool = False - self._disable_checker: bool = True + self._disable_checker: bool = False self._cloud_mobile_id_or_ref: str | None = None self._concurrency_mode: str = "per_device" self._max_concurrency: int | None = None diff --git a/artemis/sdk/types/agent.py b/artemis/sdk/types/agent.py index 1faa125..b823bdd 100644 --- a/artemis/sdk/types/agent.py +++ b/artemis/sdk/types/agent.py @@ -111,8 +111,8 @@ class AgentConfig(BaseModel): graph_config_callbacks: Callbacks = None video_recording_tools_enabled: bool = Field(default_factory=detect_video_tools_enabled) force_web_accessibility: bool = False - disable_checker: bool = True - disable_midway_checks: bool = False + disable_checker: bool = False + disable_midway_checks: bool = True disable_final_check: bool = False checker_max_iterations: int = 20 final_check_max_attempts: int = 3 @@ -122,7 +122,7 @@ class AgentConfig(BaseModel): settlement_timeout: float = 120.0 assert_failure_policy: Literal["continue", "halt"] = "continue" disable_device_probes: bool = False - disable_planner_validation: bool = True + disable_planner_validation: bool = False planner_validation_threshold: float = 0.85 enable_committee: bool = False committee_debate_rounds: int = 2 diff --git a/artemis/utils/plan_grammar.py b/artemis/utils/plan_grammar.py index 544a73b..15fc13f 100644 --- a/artemis/utils/plan_grammar.py +++ b/artemis/utils/plan_grammar.py @@ -62,6 +62,15 @@ CHECK_LINE_RE = re.compile( r"^(?P\s*)-\s*(?Pverify|assert)(?P@end)?\s*:\s*(?P\S.*?)\s*$" ) +#: Finding lines are SYSTEM-authored standing headlines for unresolved verify +#: failures, indented under the affected subgoal. They are a single-direction +#: projection of the checkpoint repair state: harness code re-renders them on +#: every plan write and NEVER reads their text back. They match neither +#: :data:`CHECKBOX_LINE_RE` nor :data:`CHECK_LINE_RE`, so they can never +#: contribute to milestone hashing, ratchet drift detection, check-item +#: multisets, or completion triggers. +FINDING_LINE_RE = re.compile(r"^(?P\s*)-\s*finding\s*:\s*(?P.*)$") + def subgoal_hash(text: str) -> str: """Stable identity hash for a subgoal text (used for per-subgoal artifacts).""" @@ -305,6 +314,58 @@ def restore_missing_check_items(content_before: str, content_after: str) -> str return "\n".join(merged) + ("\n" if (content_after or "").endswith("\n") else "") +def render_finding_line(text: str, indent: int = 2) -> str: + return f"{' ' * indent}- finding: {text}" + + +def strip_finding_lines(content: str) -> str: + """Removes every finding line (system channel hygiene before re-projection).""" + if not content: + return content + lines = [line for line in content.splitlines() if not FINDING_LINE_RE.match(line)] + return "\n".join(lines) + ("\n" if content.endswith("\n") else "") + + +def apply_finding_lines(content: str, findings_by_parent_key: dict[str, str]) -> str: + """Deterministic single-direction projection of unresolved verify findings. + + Removes ALL existing ``- finding:`` lines, then re-inserts one rendered + line per entry directly under its parent top-level subgoal (matched by + content hash). Entries whose parent subgoal no longer exists in the plan + are dropped silently — a rewritten subgoal supersedes its stale finding. + The projection is content-driven and idempotent: the model deleting a + finding line simply makes it grow back on the next write, and harness code + never reads the rendered text back. + """ + stripped = strip_finding_lines(content or "") + if not findings_by_parent_key: + return stripped + + snapshot = parse_plan(stripped) + line_by_key: dict[str, int] = {} + for item in snapshot.items: + if item.is_top_level: + line_by_key.setdefault(item.key, item.line_no) + + inserts: dict[int, list[str]] = {} + for parent_key, text in findings_by_parent_key.items(): + line_no = line_by_key.get(parent_key) + if line_no is None: + continue + inserts.setdefault(line_no, []).append(render_finding_line(text)) + + if not inserts: + return stripped + + lines = stripped.splitlines() + merged: list[str] = [] + for line_no, line in enumerate(lines): + merged.append(line) + if line_no in inserts: + merged.extend(inserts[line_no]) + return "\n".join(merged) + ("\n" if stripped.endswith("\n") else "") + + def new_top_level_completions(before: PlanSnapshot, after: PlanSnapshot) -> list[str]: """Top-level texts newly marked done, in plan order.""" before_done = {i.text for i in before.top_level if i.is_done} @@ -343,7 +404,8 @@ A top-level milestone may carry indented check lines, and the plan may end with - ` - verify: ` — an acceptance criterion for its parent milestone. It is judged by an independent Checker at the moment the milestone is marked completed; a failed verify triggers a repair loop. - ` - assert: ` — a test assertion. It is judged by the independent Checker and a failure is recorded verbatim as a legitimate test result; assertions are NEVER repaired, worked around, or satisfied by constructing state. - `- verify@end: ...` / `- assert@end: ...` — the `@end` suffix defers judgment to task exit, using the final device state. Items without `@end` are judged with the evidence available when their anchored milestone completes. -- Capability boundary (declare honestly, never promise more): all checks are POST-HOC audits over recorded evidence. They detect and record violations after the fact; they CANNOT block an action from happening, so "B must not run before A is confirmed" ordering rules are verified retroactively, not enforced. Transient prompts that were never captured in the execution history cannot be recovered — prefer expressing expectations as persistently probeable state.""" +- Capability boundary (declare honestly, never promise more): all checks are POST-HOC audits over recorded evidence. They detect and record violations after the fact; they CANNOT block an action from happening, so "B must not run before A is confirmed" ordering rules are verified retroactively, not enforced. Transient prompts that were never captured in the execution history cannot be recovered — prefer expressing expectations as persistently probeable state. +- ` - finding: ...` — a SYSTEM-authored standing headline pinned under a milestone whose verify criterion failed. You never author these lines: the system re-renders them on every plan write (deleting one only makes it reappear) and removes them automatically once the verify passes again or its repair budget is exhausted. Each finding names a `checker-...` note holding the full details and repair log — read it with `read_note`; note keys with the `checker-` prefix are reserved for the system and are read-only for you.""" def render_plan_grammar_spec(include_checks: bool) -> str: diff --git a/config/artemis.jsonc b/config/artemis.jsonc index 874fc73..f51db13 100644 --- a/config/artemis.jsonc +++ b/config/artemis.jsonc @@ -159,7 +159,7 @@ }, // 🎯 Planner Validation: Async verification when task plan milestones change "planner_validation": { - "enabled": false, + "enabled": true, // On by default: ratchet-baseline milestone validation "similarity_threshold": 0.85 }, // 🏛️ Committee: Multi-Agent Council debate tool for Operator @@ -169,9 +169,9 @@ }, // 🔎 Checker: Plan-declared checkpoint verification (optimistic async) + exit final review "checker": { - "enabled": false, // Master switch (compat alias): false disables both gates below - "midway_checks": true, // Plan-declared midway checkpoints (verify/assert on subgoal completion) - "final_check": true, // Exit final review against the user's original goal + "enabled": true, // Master switch (compat alias): false disables both gates below + "midway_checks": false, // Plan-declared midway checkpoints (off in the factory layering) + "final_check": true, // Exit final review against the user's original goal (on by default) "max_iterations": 20, "final_check_max_attempts": 3, "checkpoint_max_repairs": 2, diff --git a/tests/unit/agents/test_planner_node.py b/tests/unit/agents/test_planner_node.py index 2501529..9fabb11 100644 --- a/tests/unit/agents/test_planner_node.py +++ b/tests/unit/agents/test_planner_node.py @@ -84,3 +84,29 @@ async def test_planner_initial_plan(mock_context): assert isinstance(human_msg_content, list) assert human_msg_content[0]["type"] == "text" assert "Goal: Test Goal" in human_msg_content[0]["text"] + + +def test_validate_plan_format_single_sourced_status_alphabet(): + """Regression: the Planner's format gate is single-sourced from + plan_grammar — a legal in-progress '[/]' line must never be rejected.""" + from artemis.agents.planner.planner import validate_plan_format + + valid, err = validate_plan_format( + "- [x] Done milestone\n- [/] In-progress milestone\n - [ ] Sub\n- [!] Blocked\n" + ) + assert valid is True and err == "" + + # Every grammar status char is accepted; unknown ones are still rejected. + from artemis.utils.plan_grammar import STATUS_CHARS + + for c in STATUS_CHARS: + ok, _ = validate_plan_format(f"- [{c}] Milestone\n") + assert ok is True + + bad, message = validate_plan_format("- [z] Bogus status\n") + assert bad is False + assert "Invalid task status" in message + + empty, message = validate_plan_format("# just prose\n") + assert empty is False + assert "at least one subgoal" in message diff --git a/tests/unit/graph/test_checkpoints.py b/tests/unit/graph/test_checkpoints.py index d3be08d..77c6257 100644 --- a/tests/unit/graph/test_checkpoints.py +++ b/tests/unit/graph/test_checkpoints.py @@ -27,13 +27,15 @@ from artemis.context import ArtemisContext, ExecutionSetup from artemis.graph.checkpoints import ( CheckpointRun, PendingCheckpoint, + checker_note_key, harvest_finished_checkpoints, harvest_run, + is_checker_note_key, queue_checkpoints, read_ledger, spawn_pending_checkpoints, ) -from artemis.graph.graph import execution_check_node, wrap_note_tool +from artemis.graph.graph import execution_check_node, wrap_note_tool, wrap_update_note_tool from artemis.graph.state import State from artemis.utils.plan_grammar import CheckItem, parse_plan, subgoal_hash @@ -50,6 +52,9 @@ PLAN_WITH_CHECKS = ( def _make_ctx(tmp_path, **setup_kwargs): setup_kwargs.setdefault("disable_checker", False) + # These tests exercise the midway checkpoint machinery, which is off in + # the factory layering — pin it on unless a test overrides it. + setup_kwargs.setdefault("disable_midway_checks", False) ctx = MagicMock(spec=ArtemisContext) ctx.execution_setup = ExecutionSetup(**setup_kwargs) ctx.data_engine = MagicMock() @@ -528,6 +533,150 @@ async def test_timeout_records_inconclusive_not_passed(tmp_path): assert all(r["status"] != "passed" for r in records) +# --- Verify-finding four-layer persistence ------------------------------------------- + + +def _checker_note_path(tmp_path): + return tmp_path / "notes" / f"{checker_note_key(GOAL_KEY)}.md" + + +@pytest.mark.asyncio +async def test_verify_fail_writes_finding_line_and_checker_note(tmp_path): + plan_path = _write_plan(tmp_path) + ctx = _make_ctx(tmp_path) + run = _done_run(ctx, CheckReport(verdicts=[_verdict("verify", "failed")])) + + harvest_run(ctx, _make_state(), run, allow_side_effects=True) + + content = plan_path.read_text(encoding="utf-8") + lines = content.splitlines() + # Standing headline pinned directly under the (now reverted) subgoal, + # naming the failed criterion and pointing at the detail note. + idx = lines.index(f"- [/] {GOAL_TEXT}") + assert lines[idx + 1].startswith(" - finding: verify failed") + assert checker_note_key(GOAL_KEY) in lines[idx + 1] + + # Layer 3: the system note carries the full criterion/evidence/suggestion. + note = _checker_note_path(tmp_path).read_text(encoding="utf-8") + assert "the alarm list shows 7:30 AM" in note + assert "concrete evidence" in note + assert "fix it" in note + + +@pytest.mark.asyncio +async def test_finding_line_regrows_on_model_plan_write(tmp_path): + plan_path = _write_plan(tmp_path) + ctx = _make_ctx(tmp_path) + run = _done_run(ctx, CheckReport(verdicts=[_verdict("verify", "failed")])) + harvest_run(ctx, _make_state(), run, allow_side_effects=True) + assert "- finding:" in plan_path.read_text(encoding="utf-8") + + # The model rewrites the plan WITHOUT the finding line... + clean_rewrite = PLAN_WITH_CHECKS.replace(f"- [x] {GOAL_TEXT}", f"- [/] {GOAL_TEXT}") + + async def fake_invoke(tool, args, tool_call_id, state, record_trace=None): + plan_path.write_text(args["content"], encoding="utf-8") + return "Success" + + mock_tool = MagicMock() + mock_tool.name = "save_note" + mock_tool.description = "save" + + with patch("artemis.graph.graph.invoke_tool_with_injection", side_effect=fake_invoke): + wrapped = wrap_note_tool(ctx, mock_tool) + await wrapped.ainvoke( + {"key": "task_plan", "content": clean_rewrite, "tool_call_id": "t1"} + ) + + # ...and the deterministic projection grows it back on the write path. + assert "- finding: verify failed" in plan_path.read_text(encoding="utf-8") + + +@pytest.mark.asyncio +async def test_finding_line_removed_after_verify_passes_note_kept(tmp_path): + plan_path = _write_plan(tmp_path) + ctx = _make_ctx(tmp_path) + run1 = _done_run(ctx, CheckReport(verdicts=[_verdict("verify", "failed")])) + harvest_run(ctx, _make_state(), run1, allow_side_effects=True) + assert "- finding:" in plan_path.read_text(encoding="utf-8") + + run2 = _done_run( + ctx, + CheckReport(verdicts=[_verdict("verify", "passed")]), + attempt_id=f"{GOAL_KEY}#2", + ) + harvest_run(ctx, _make_state(), run2, allow_side_effects=True) + + # Layer 2 headline retired; layer 3 note stays on disk with the full + # per-attempt repair log (fail then pass). + assert "- finding:" not in plan_path.read_text(encoding="utf-8") + note = _checker_note_path(tmp_path).read_text(encoding="utf-8") + assert f"attempt {GOAL_KEY}#1 — verify failed" in note + assert f"attempt {GOAL_KEY}#2 — verify passed" in note + + +@pytest.mark.asyncio +async def test_finding_line_removed_on_repair_quota_exhaustion(tmp_path): + plan_path = _write_plan(tmp_path) + ctx = _make_ctx(tmp_path, checkpoint_max_repairs=1) + run1 = _done_run(ctx, CheckReport(verdicts=[_verdict("verify", "failed")])) + harvest_run(ctx, _make_state(), run1, allow_side_effects=True) + assert "- finding:" in plan_path.read_text(encoding="utf-8") + + # Re-complete, fail again: quota (1) is exhausted -> finding settled. + plan_path.write_text( + plan_path.read_text(encoding="utf-8").replace( + f"- [/] {GOAL_TEXT}", f"- [x] {GOAL_TEXT}" + ), + encoding="utf-8", + ) + run2 = _done_run( + ctx, + CheckReport(verdicts=[_verdict("verify", "failed")]), + attempt_id=f"{GOAL_KEY}#2", + ) + harvest_run(ctx, _make_state(), run2, allow_side_effects=True) + + content = plan_path.read_text(encoding="utf-8") + assert "- finding:" not in content + assert f"- [x] {GOAL_TEXT}" in content # no further revert either + + +@pytest.mark.asyncio +async def test_checker_note_prefix_writes_rejected_by_wrappers(tmp_path): + ctx = _make_ctx(tmp_path) + save_tool = MagicMock() + save_tool.name = "save_note" + save_tool.description = "save" + update_tool = MagicMock() + update_tool.name = "update_note" + update_tool.description = "update" + + with patch( + "artemis.graph.graph.invoke_tool_with_injection", new_callable=AsyncMock + ) as inv: + wrapped_save = wrap_note_tool(ctx, save_tool) + result = await wrapped_save.ainvoke( + {"key": checker_note_key(GOAL_KEY), "content": "spoof", "tool_call_id": "t1"} + ) + assert "reserved" in str(result.content) + + wrapped_update = wrap_update_note_tool(ctx, update_tool) + result2 = await wrapped_update.ainvoke( + { + "key": "checker:abc", + "target": "a", + "replacement": "b", + "tool_call_id": "t2", + } + ) + assert "reserved" in str(result2.content) + + inv.assert_not_awaited() + assert is_checker_note_key("checker-123") and is_checker_note_key("checker:123") + assert not is_checker_note_key("checkerboard_notes") and not is_checker_note_key("notes") + + @pytest.mark.asyncio async def test_exception_fail_open_releases_but_books_inconclusive(tmp_path): plan_path = _write_plan(tmp_path) diff --git a/tests/unit/test_config_package.py b/tests/unit/test_config_package.py index f1c4295..8f3b187 100644 --- a/tests/unit/test_config_package.py +++ b/tests/unit/test_config_package.py @@ -179,13 +179,13 @@ def test_agent_config_loading(): assert agent_cfg.explorer.caching is True assert "explorer" in agent_cfg.denylisted_tools assert agent_cfg.video_analyzer.enable_ledger is True - assert agent_cfg.planner_validation.enabled is False + assert agent_cfg.planner_validation.enabled is True assert agent_cfg.planner_validation.similarity_threshold == 0.85 assert agent_cfg.committee.enabled is False assert agent_cfg.committee.debate_rounds == 2 - assert agent_cfg.checker.enabled is False + assert agent_cfg.checker.enabled is True assert agent_cfg.checker.max_iterations == 20 - assert agent_cfg.checker.midway_checks is True + assert agent_cfg.checker.midway_checks is False assert agent_cfg.checker.final_check is True assert agent_cfg.checker.assert_failure_policy == "continue" assert agent_cfg.checker.device_probes is True @@ -194,9 +194,9 @@ def test_agent_config_loading(): assert agent_cfg.flash.max_turns == 30 assert agent_cfg.flash.explorer_mode == "flash" assert agent_cfg.pro.explorer.mode == "flash" - assert agent_cfg.pro.checker.enabled is False + assert agent_cfg.pro.checker.enabled is True assert agent_cfg.pro.committee.enabled is False - assert agent_cfg.pro.planner_validation.enabled is False + assert agent_cfg.pro.planner_validation.enabled is True assert agent_cfg.pro.video_analyzer.enable_ledger is True @@ -255,10 +255,10 @@ def test_planner_validation_builder_and_milestones(): from artemis.sdk.builders.agent_config_builder import AgentConfigBuilder from artemis.utils.plan_grammar import milestones_changed, parse_plan - # Default builder inherits from artemis.jsonc (enabled=False, similarity_threshold=0.85) + # Default builder inherits from artemis.jsonc (enabled=True, similarity_threshold=0.85) builder = AgentConfigBuilder() cfg = builder.build() - assert cfg.disable_planner_validation is True + assert cfg.disable_planner_validation is False assert cfg.planner_validation_threshold == 0.85 # Fluent enabling @@ -338,12 +338,12 @@ def test_checker_builder_and_context_propagation(): from artemis.sdk.agent import Agent from artemis.sdk.builders.agent_config_builder import AgentConfigBuilder - # Default builder inherits from artemis.jsonc (enabled=False, max_iterations=20) + # Default builder inherits from artemis.jsonc (enabled=True, midway off, final on) builder = AgentConfigBuilder() cfg = builder.build() - assert cfg.disable_checker is True + assert cfg.disable_checker is False assert cfg.checker_max_iterations == 20 - assert cfg.disable_midway_checks is False + assert cfg.disable_midway_checks is True assert cfg.disable_final_check is False # Fluent enabling with the new two-gate knobs @@ -402,6 +402,47 @@ def test_checker_builder_and_context_propagation(): assert ctx.execution_setup.final_check_enabled is False +def test_factory_default_verification_layering(): + """Contract: out of the box, the verification stack is layered as + final check ON / planner validation (ratchet) ON / midway checks OFF.""" + from artemis.config.agent import AgentGlobalConfig, CheckerConfig, PlannerValidationConfig + from artemis.context import ExecutionSetup + from artemis.sdk.builders.agent_config_builder import AgentConfigBuilder + + # Config-model factory defaults + assert CheckerConfig().enabled is True + assert CheckerConfig().midway_checks is False + assert CheckerConfig().final_check is True + assert PlannerValidationConfig().enabled is True + + # Bare ExecutionSetup carries the same layering, including the effective + # gate semantics (master on + midway off => final review runs, no midway + # checkpoint ever spawns). + setup = ExecutionSetup() + assert setup.disable_checker is False + assert setup.disable_midway_checks is True + assert setup.disable_final_check is False + assert setup.disable_planner_validation is False + assert setup.final_check_enabled is True + assert setup.midway_checks_enabled is False + assert setup.checks_enabled is True + + # Builder default (fed by config/artemis.jsonc) agrees + cfg = AgentConfigBuilder().build() + assert cfg.disable_checker is False + assert cfg.disable_midway_checks is True + assert cfg.disable_final_check is False + assert cfg.disable_planner_validation is False + + # Shipped artemis.jsonc agrees with the model factory defaults + agent_cfg = load_agent_config() + assert agent_cfg.checker.enabled is True + assert agent_cfg.checker.midway_checks is False + assert agent_cfg.checker.final_check is True + assert agent_cfg.planner_validation.enabled is True + assert AgentGlobalConfig().checker.enabled is True + + def test_explorer_builder_and_resolution(monkeypatch): """Test AgentConfigBuilder explorer methods and multi-tier resolution logic.""" from artemis.context import ArtemisContext, DeviceContext, DevicePlatform diff --git a/tests/unit/utils/test_plan_grammar.py b/tests/unit/utils/test_plan_grammar.py index a6b0a43..f45a0a1 100644 --- a/tests/unit/utils/test_plan_grammar.py +++ b/tests/unit/utils/test_plan_grammar.py @@ -239,3 +239,59 @@ def test_prompts_carry_the_single_sourced_grammar(): planner = json.loads(planner_raw) guidelines = planner["blocks"]["validator_guidelines"] assert "until manually stopped" not in guidelines + + +# --- Finding lines (system channel; single-direction projection) --------------------- + + +def test_finding_lines_are_invisible_to_the_machine_channel(): + from artemis.utils.plan_grammar import apply_finding_lines + + plan = "- [x] G1\n - verify: V1\n- [ ] G2\n" + key = subgoal_hash("G1") + with_finding = apply_finding_lines( + plan, {key: "verify failed — 'V1' (details: note 'checker-x')"} + ) + + assert "- finding: verify failed" in with_finding + + before, after = parse_plan(plan), parse_plan(with_finding) + # Same items, same hashes, same check items: no milestone drift, no + # planner-validation trigger, no check multiset change. + assert [i.key for i in after.items] == [i.key for i in before.items] + assert milestones_changed(before, after) is False + assert check_items_changed(before, after) is False + assert new_top_level_completions(before, after) == [] + assert unintended_milestone_edits(before, after) == [] + + +def test_apply_finding_lines_projection_is_idempotent_and_anchored(): + from artemis.utils.plan_grammar import apply_finding_lines, strip_finding_lines + + plan = "- [x] G1\n - verify: V1\n- [/] G2\n" + findings = {subgoal_hash("G1"): "headline one"} + once = apply_finding_lines(plan, findings) + twice = apply_finding_lines(once, findings) + assert once == twice # idempotent projection + # Anchored directly under its parent subgoal + lines = once.splitlines() + assert lines[lines.index("- [x] G1") + 1] == " - finding: headline one" + + # Resolution: an empty registry strips the line back out + assert apply_finding_lines(once, {}) == plan + assert strip_finding_lines(once) == plan + + +def test_apply_finding_lines_drops_entries_for_missing_subgoals(): + from artemis.utils.plan_grammar import apply_finding_lines + + plan = "- [x] G1 (rephrased)\n" + projected = apply_finding_lines(plan, {subgoal_hash("G1"): "stale headline"}) + assert "finding" not in projected + + +def test_finding_grammar_doc_is_conditional(): + base = render_plan_grammar_spec(include_checks=False) + extended = render_plan_grammar_spec(include_checks=True) + assert "finding" not in base + assert "- finding:" in extended and "checker-" in extended