Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 23 additions & 6 deletions anton/core/llm/prompt_builder.py
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,18 @@ def _build_tool_prompts_section(self, tool_defs: list["ToolDef"] | None) -> str:

return "\n\n".join(chunks)

# Built-in skills that are always available regardless of ~/.anton/skills/.
# Each entry must have "label" and "when_to_use".
_BUILTIN_SKILL_SUMMARIES: list[dict] = [
{
"label": "generate_dashboard_html",
"when_to_use": (
"when the user asks to build a dashboard, chart, report, "
"presentation, or any data visualization as a standalone HTML page"
),
},
]

def _build_procedural_memory_section(
self, skill_store: "SkillStore | None"
) -> str:
Expand All @@ -71,12 +83,17 @@ def _build_procedural_memory_section(
the full procedure. Returns an empty string if no store is wired
or no skills are saved — the caller skips the section entirely.
"""
if skill_store is None:
return ""
try:
summaries = skill_store.list_summaries()
except Exception:
return ""
summaries: list[dict] = list(self._BUILTIN_SKILL_SUMMARIES)
if skill_store is not None:
try:
user_summaries = skill_store.list_summaries()
except Exception:
user_summaries = []
# User skills override builtins with the same label.
builtin_labels = {s["label"] for s in summaries}
for s in user_summaries:
if s.get("label") not in builtin_labels:
summaries.append(s)
if not summaries:
return ""

Expand Down
464 changes: 349 additions & 115 deletions anton/core/llm/prompts.py

Large diffs are not rendered by default.

6 changes: 6 additions & 0 deletions anton/core/session.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
from anton.core.memory.cerebellum import Cerebellum
from anton.core.memory.skills import SkillStore
from anton.core.tools.recall_skill import RECALL_SKILL_TOOL
from anton.core.tools.erase_scratchpad_history import ERASE_SCRATCHPAD_HISTORY_TOOL
from anton.core.llm.prompts import RESILIENCE_NUDGE
from anton.core.llm.provider import (
ContextOverflowError,
Expand Down Expand Up @@ -411,6 +412,11 @@ def _build_core_tools(self) -> None:
# Procedural memory retrieval — always available, no-op if no skills.
self.tool_registry.register_tool(RECALL_SKILL_TOOL)

# Context compaction for file-artifact workflows: the agent marks
# intermediate cells with `# DELETABLE: <desc>` and calls this after
# writing the final file to disk.
self.tool_registry.register_tool(ERASE_SCRATCHPAD_HISTORY_TOOL)

async def close(self) -> None:
"""Clean up scratchpads and other resources."""
await self._scratchpads.close_all()
Expand Down
146 changes: 146 additions & 0 deletions anton/core/tools/erase_scratchpad_history.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,146 @@
"""The `erase_scratchpad_history` tool — drop intermediate scratchpad code from
context once the artifact it produced has been written to disk.

When the agent builds an HTML dashboard (or any file artifact) across many
scratchpad cells, every `exec` call and its code string lives on in the message
history. Once the final HTML is on disk, that intermediate code is redundant —
the file is the source of truth. This tool blanks the code of cells that the
agent marked with `# DELETABLE: <description>` as the first line, both in the
live pad and in the assistant message history, so follow-up turns don't pay
the context tax.

Scope is deliberately narrow: only `input.code` of matching scratchpad exec
tool calls and `cell.code` of matching cells. Outputs, descriptions, and other
tool types are untouched.
"""

from __future__ import annotations

from typing import TYPE_CHECKING

from anton.core.tools.tool_defs import ToolDef

if TYPE_CHECKING:
from anton.core.session import ChatSession


_MARKER = "# DELETABLE:"


_DESCRIPTION = (
"Erase the Python code of scratchpad cells you previously marked with "
"`# DELETABLE: <short description>` as their first line. Call this after "
"a file artifact (HTML dashboard, presentation, etc.) has been written to "
"disk — the file is the source of truth, so the intermediate generation "
"code is no longer useful in the conversation context.\n\n"
"The tool scans both the live scratchpad cells and the assistant message "
"history, and replaces every matching `code` value with "
"`# DELETED: <description>` (the description from the marker is preserved "
"so you can still see what each cell did). Outputs, descriptions on the "
"tool call, and anything without the marker are left untouched. Safe to "
"call repeatedly — cells already cleared are skipped."
)


_INPUT_SCHEMA = {
"type": "object",
"properties": {},
"required": [],
"additionalProperties": False,
}


def _extract_deletable_description(code: str) -> str | None:
"""If the first non-blank line of `code` is `# DELETABLE: <desc>`, return
<desc>. Otherwise None.

Leading blank lines (common when the model emits multi-line code blocks)
are ignored so the marker still matches.
"""
if not isinstance(code, str):
return None
stripped = code.lstrip()
first_line = stripped.split("\n", 1)[0].strip()
if not first_line.startswith(_MARKER):
return None
return first_line[len(_MARKER):].strip()


def _replacement(description: str) -> str:
if description:
return f"# DELETED: {description}"
return "# DELETED"


async def handle_erase_scratchpad_history(
session: "ChatSession", tc_input: dict
) -> str:
"""Clear code of DELETABLE-marked cells in live pads and history."""
cells_cleared = 0
history_blocks_cleared = 0

# 1. Live scratchpad cells
for pad in session._scratchpads.pads.values():
for cell in pad.cells:
desc = _extract_deletable_description(cell.code)
if desc is None:
continue
cell.code = _replacement(desc)
cells_cleared += 1

# 2. Message history — only scratchpad exec tool_use blocks.
# Skip the last assistant message (it contains this tool call itself).
history = session._history
last_idx = len(history) - 1
for idx, msg in enumerate(history):
if idx == last_idx and msg.get("role") == "assistant":
continue
content = msg.get("content")
if not isinstance(content, list):
continue
for block in content:
if not isinstance(block, dict):
continue
if block.get("type") != "tool_use":
continue
if block.get("name") != "scratchpad":
continue
tc_in = block.get("input")
if not isinstance(tc_in, dict):
continue
if tc_in.get("action") != "exec":
continue
code = tc_in.get("code")
desc = _extract_deletable_description(code)
if desc is None:
continue
tc_in["code"] = _replacement(desc)
history_blocks_cleared += 1

session._persist_history()

if cells_cleared == 0 and history_blocks_cleared == 0:
return (
"No DELETABLE cells found. Either none were marked with "
"`# DELETABLE: <description>` on their first line, or they have "
"already been cleared."
)

return (
f"Cleared {cells_cleared} live scratchpad cells and "
f"{history_blocks_cleared} tool-call entries in history."
)


ERASE_SCRATCHPAD_HISTORY_TOOL = ToolDef(
name="erase_scratchpad_history",
description=_DESCRIPTION,
input_schema=_INPUT_SCHEMA,
handler=handle_erase_scratchpad_history,
)


__all__ = [
"ERASE_SCRATCHPAD_HISTORY_TOOL",
"handle_erase_scratchpad_history",
]
27 changes: 27 additions & 0 deletions anton/core/tools/recall_skill.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,10 +14,19 @@

from __future__ import annotations

from pathlib import Path
from typing import TYPE_CHECKING

from anton.core.llm.prompts import VISUALIZATIONS_HTML_OUTPUT_FORMAT_PROMPT
from anton.core.tools.tool_defs import ToolDef

_DASHBOARD_TEMPLATE_PATH = (
Path(__file__).resolve().parent.parent.parent
/ "templates"
/ "template-dark.html"
)
_DASHBOARD_TEMPLATE_MARKER = "<!-- DASHBOARD_HTML_TEMPLATE -->"

if TYPE_CHECKING:
from anton.core.session import ChatSession

Expand Down Expand Up @@ -51,6 +60,21 @@
}


def get_generate_dashboard_html_prompt(
session: "ChatSession | None" = None,
) -> str:
output_context = ""
if session is not None:
ctx = getattr(session, "_system_prompt_context", None)
if ctx is not None:
output_context = getattr(ctx, "output_context", "") or ""
rendered = VISUALIZATIONS_HTML_OUTPUT_FORMAT_PROMPT.format(
output_context=output_context
)
html_template = _DASHBOARD_TEMPLATE_PATH.read_text(encoding="utf-8")
return rendered.replace(_DASHBOARD_TEMPLATE_MARKER, html_template)


def _format_skill_response(skill, *, warning: str = "") -> str:
"""Render the recall payload sent back to the LLM as a tool result."""
parts: list[str] = []
Expand All @@ -77,6 +101,9 @@ async def handle_recall_skill(session: "ChatSession", tc_input: dict) -> str:
"Pick one from the procedural memory list in your system prompt."
)

if label_in == "generate_dashboard_html":
return get_generate_dashboard_html_prompt(session)

store = getattr(session, "_skill_store", None)
if store is None:
return (
Expand Down
Loading
Loading