Staging (Week of 26-07-13)

anton/README.md

@@ -305,7 +305,7 @@ When scratchpads are active, relevant lessons are appended to the scratchpad too
 scratchpad_tool["description"] += f"\n\nLessons from past sessions:\n{wisdom}"
 ```
 
-This combines all "when" rules + lessons with `scratchpad-*` topics from both scopes. The content comes from `cortex.get_scratchpad_context()`, which calls `recall_scratchpad_wisdom()` on both hippocampi.
+This combines scratchpad-related "when" rules + scratchpad-related lessons from both scopes, ordered by confidence tier (high, then medium/unset, then low) and recency within a tier, trimmed to a token budget (default 2000/scope, ~4 chars/token). The content comes from `cortex.get_scratchpad_context()`, which calls `recall_scratchpad_wisdom()` on both hippocampi.
 
 ## The `memorize` Tool
 
@@ -572,7 +572,7 @@ When errored cells exist, they accumulate in a buffer across the turn. At end-of
 1. The buffered cells get formatted into a compact post-mortem prompt
 2. One LLM call via `LLMClient.generate_object_code` (the cheap coding model) returns a `_DiffPassResult` Pydantic model with extracted lessons
 3. Each lesson is wrapped as an `Engram` with `kind="lesson"`, `topic="scratchpad"`, `source="consolidation"`, and routed through `Cortex.encode()` — the same path manual lessons and the consolidator already use
-4. Future scratchpad cells see those lessons via the existing `recall_scratchpad_wisdom()` injection into the scratchpad tool description
+4. Future scratchpad cells see those lessons via the existing `recall_scratchpad_wisdom()` injection into the scratchpad tool description (confidence tier + recency ordered, budget-limited — not all lessons are guaranteed to survive the cut)
 
 The cerebellum is a **producer** only — it generates new lesson entries for the existing storage and retrieval pipeline. There's no parallel storage system, no separate `corrections.md` file. Whatever the consolidator and `/memorize` write to, the cerebellum also writes to.
 
@@ -811,7 +811,7 @@ The Hippocampus handles one scope (global OR project) and is the canonical file-
 | `recall_rules()` | `rules.md` | Basal Ganglia + OFC |
 | `recall_lessons(token_budget)` | `lessons.md` (budget-limited, most recent first) | Anterior Temporal Lobe |
 | `recall_topic(slug)` | `topics/{slug}.md` | Cortical Association Areas |
-| `recall_scratchpad_wisdom()` | "when" rules + scratchpad-related lessons + `topics/scratchpad-*.md` | Procedural memory |
+| `recall_scratchpad_wisdom(token_budget)` | scratchpad-related "when" rules + scratchpad-related lessons (confidence tier + recency ordered, budget-limited) | Procedural memory |
 
 **Encoding methods:**
 | Method | Writes | Behavior |

anton/core/memory/base.py

@@ -53,16 +53,21 @@ def recall_rules(self) -> str:
         """Return behavioral gates (rules.md equivalent)."""
         ...
 
-    def recall_lessons(self, token_budget: int = 1000) -> str:
-        """Return semantic facts within the given token budget."""
+    def recall_lessons(self, token_budget: int = 1000, exclude_scratchpad: bool = False) -> str:
+        """Return semantic facts within the given token budget.
+
+        exclude_scratchpad drops entries whose text/topic mentions
+        "scratchpad" — used by build_memory_context() to avoid double-billing
+        entries already injected via recall_scratchpad_wisdom().
+        """
         ...
 
     def recall_topic(self, slug: str) -> str:
         """Return lessons tagged with the given topic slug."""
         ...
 
-    def recall_scratchpad_wisdom(self) -> str:
-        """Return procedural knowledge relevant to scratchpad execution."""
+    def recall_scratchpad_wisdom(self, token_budget: int = 2000) -> str:
+        """Return procedural knowledge relevant to scratchpad execution, within budget."""
         ...
 
     # --- read (Engrams, for inspection / CRUD UI) ---
@@ -71,12 +76,22 @@ def get_identities(self) -> list[Engram]:
         """Return identity entries as Engrams."""
         ...
 
-    def get_rules(self) -> list[Engram]:
-        """Return behavioral rules as Engrams."""
+    def get_rules(self, exclude_scratchpad_when: bool = False) -> list[Engram]:
+        """Return behavioral rules as Engrams.
+
+        exclude_scratchpad_when drops "when" rules whose text/topic mentions
+        "scratchpad" — used by build_memory_context() to avoid double-billing
+        entries already injected via recall_scratchpad_wisdom(). always/never
+        rules are never affected.
+        """
         ...
 
-    def get_lessons(self, token_budget: int = None) -> list[Engram]:
-        """Return semantic facts as Engrams, optionally budget-limited."""
+    def get_lessons(self, token_budget: int = None, exclude_scratchpad: bool = False) -> list[Engram]:
+        """Return semantic facts as Engrams, optionally budget-limited.
+
+        exclude_scratchpad drops entries whose text/topic mentions
+        "scratchpad", applied before the budget cut.
+        """
         ...
 
     # --- write ---

anton/core/memory/cortex.py

@@ -185,17 +185,20 @@ async def build_memory_context(self, user_message: str = "") -> str:
         if identity:
             sections.append(f"## Your Memory — Identity\n{identity}")
 
-        # 2. Global rules (with smart retrieval)
-        global_engrams = self.global_hc.get_rules()
+        # 2. Global rules (with smart retrieval). get_rules(exclude_scratchpad_when=True)
+        # drops scratchpad-related "when" rules — those are already injected
+        # into the scratchpad tool description by get_scratchpad_context(),
+        # and showing them here too would double their token cost.
+        global_engrams = self.global_hc.get_rules(exclude_scratchpad_when=True)
         if global_engrams:
             global_engrams = await self._retrieve_relevant_rules(global_engrams, user_message)
             if global_engrams:
                 sections.append(
                     f"## Your Memory — Global Rules\n{self._format_rules_engrams(global_engrams)}"
                 )
 
-        # 3. Project rules (with smart retrieval)
-        project_engrams = self.project_hc.get_rules()
+        # 3. Project rules (with smart retrieval) — same scratchpad exclusion.
+        project_engrams = self.project_hc.get_rules(exclude_scratchpad_when=True)
         if project_engrams:
             project_engrams = await self._retrieve_relevant_rules(project_engrams, user_message)
             if project_engrams:
@@ -205,17 +208,18 @@ async def build_memory_context(self, user_message: str = "") -> str:
                 for engram in project_engrams:
                     self._log_read_engram(engram)
 
-        # 4. Global lessons
-        global_lessons = self.global_hc.recall_lessons(token_budget=1000)
+        # 4. Global lessons. recall_lessons() excludes scratchpad-related
+        # entries internally — same reasoning as the rules exclusion above.
+        global_lessons = self.global_hc.recall_lessons(token_budget=1000, exclude_scratchpad=True)
         if global_lessons:
             sections.append(f"## Your Memory — Global Lessons\n{global_lessons}")
 
-        # 5. Project lessons
-        project_lessons = self.project_hc.recall_lessons(token_budget=1000)
+        # 5. Project lessons — same scratchpad exclusion.
+        project_lessons = self.project_hc.recall_lessons(token_budget=1000, exclude_scratchpad=True)
         if project_lessons:
             sections.append(f"## Your Memory — Project Lessons\n{project_lessons}")
             if self._episodic is not None:
-                for engram in self.project_hc.get_lessons(token_budget=1000):
+                for engram in self.project_hc.get_lessons(token_budget=1000, exclude_scratchpad=True):
                     self._log_read_engram(engram)
 
         # 6. Minds datasource context (auto-loaded if present)

anton/core/memory/hippocampus.py

@@ -67,6 +67,39 @@ def _extract_metadata(text: str) -> tuple[str, dict]:
     return clean_text, meta
 
 
+def _filter_by_token_budget(engrams: list[Engram], token_budget: int) -> list[Engram]:
+    """Keep entries, in the given order, until token_budget is spent.
+
+    Callers must pass entries already sorted by priority (most important
+    first) — this stops at the first entry that would exceed the budget, it
+    does not repack for a tighter fit. Budget enforced at ~4 chars/token.
+    """
+    char_budget = token_budget * 4
+    used = 0
+    kept: list[Engram] = []
+    for engram in engrams:
+        length = len(engram.text) + 1
+        if used + length > char_budget:
+            break
+        kept.append(engram)
+        used += length
+    return kept
+
+
+def _is_scratchpad_related(engram: Engram) -> bool:
+    """Whether an engram's text or topic mentions "scratchpad".
+
+    Used both to select entries INTO recall_scratchpad_wisdom() (the
+    scratchpad tool description) and to exclude the same entries FROM
+    get_rules()/get_lessons() when building the system prompt — keeping one
+    definition avoids the two call sites drifting apart and double-injecting
+    (or double-excluding) an entry.
+    """
+    if "scratchpad" in engram.text.lower():
+        return True
+    return bool(engram.topic and "scratchpad" in engram.topic.lower())
+
+
 class Hippocampus:
     """Reads and writes memory traces at a single scope (global OR project).
 
@@ -191,16 +224,16 @@ def clear_identity(self) -> None:
 
     # ---------  lessons --------------
 
-    def recall_lessons(self, token_budget: int|None = 1000) -> str:
+    def recall_lessons(self, token_budget: int|None = 1000, exclude_scratchpad: bool = False) -> str:
         """Load semantic knowledge (lessons.md), most recent first, within budget.
 
         Brain analog: Anterior Temporal Lobe — the convergence hub for semantic
         facts distilled from many episodes. Budget enforced at ~4 chars/token.
         """
-        return self._lessons_to_text(self.get_lessons(token_budget))
+        return self._lessons_to_text(self.get_lessons(token_budget, exclude_scratchpad=exclude_scratchpad))
 
 
-    def get_lessons(self, token_budget: int = None) -> list[Engram]:
+    def get_lessons(self, token_budget: int = None, exclude_scratchpad: bool = False) -> list[Engram]:
         """Load semantic knowledge (lessons.md) as Engrams.
 
         When token_budget is None (default) returns all entries in file order.
@@ -233,24 +266,16 @@ def get_lessons(self, token_budget: int = None) -> list[Engram]:
             text, meta = _extract_metadata(text)
             entries.append(Engram(text=text.removeprefix("- "), **meta))
 
+        if exclude_scratchpad:
+            entries = [e for e in entries if not _is_scratchpad_related(e)]
+
         if token_budget is None:
             return entries
 
         # Reverse entries so most recent are first
         entries.reverse()
 
-        # Budget: ~4 chars per token
-        char_budget = token_budget * 4
-        result_lines = []
-        used = sum(len(ln) for ln in result_lines)
-
-        for ln in entries:
-            if used + len(ln.text) + 1 > char_budget:
-                break
-            result_lines.append(ln)
-            used += len(ln.text) + 1
-
-        return result_lines
+        return _filter_by_token_budget(entries, token_budget)
 
     def del_lesson(self, id):
         entries = self.get_lessons()
@@ -320,36 +345,46 @@ def recall_topic(self, slug: str) -> str:
 
         return self._lessons_to_text(items, header=slug)
 
-    def recall_scratchpad_wisdom(self) -> str:
+    def recall_scratchpad_wisdom(self, token_budget: int = 2000) -> str:
         """Retrieve procedural knowledge relevant to scratchpad execution.
 
-        Returns all "when" rules + lessons whose text contains "scratchpad".
-        Injected into tool descriptions so the LLM sees them when composing code.
+        Gathers "when" rules + lessons whose text or topic mentions
+        "scratchpad", ordered by confidence tier (high, then medium/unset,
+        then low) and by recency within a tier, then trimmed to token_budget
+        (~4 chars/token — same convention as get_lessons). Injected into tool
+        descriptions so the LLM sees them when composing code.
         """
-        parts: list[str] = []
-
-        # Extract "when" rules
-        rules = self.recall_rules()
-        if rules:
-            in_when = False
-            for line in rules.splitlines():
-                if line.strip().startswith("## When"):
-                    in_when = True
-                    continue
-                elif line.strip().startswith("## "):
-                    in_when = False
-                    continue
-                if in_when and line.strip().startswith("- "):
-                    parts.append(line.strip())
+        candidates: list[Engram] = []
+
+        for rule in self.get_rules():
+            if rule.kind == "when" and _is_scratchpad_related(rule):
+                candidates.append(rule)
 
-        # Extract scratchpad-related lessons
         for lesson in self.get_lessons():
-            if "scratchpad" in lesson.text.lower() or (lesson.topic and "scratchpad" in lesson.topic.lower()):
-                entry = f"- {lesson.text}"
-                if entry not in parts:
-                    parts.append(entry)
+            if _is_scratchpad_related(lesson):
+                candidates.append(lesson)
+
+        confidence_rank = {"high": 0, "medium": 1, "low": 2}
+
+        def sort_key(engram: Engram) -> tuple[int, float]:
+            rank = confidence_rank.get(engram.confidence, 1)
+            recency = -engram.updated_at.timestamp() if engram.updated_at else float("inf")
+            return (rank, recency)
+
+        candidates.sort(key=sort_key)
 
-        return "\n".join(parts)
+        # Dedup by text, keeping the higher-priority copy from the sort above
+        seen_texts: set[str] = set()
+        deduped: list[Engram] = []
+        for engram in candidates:
+            if engram.text in seen_texts:
+                continue
+            seen_texts.add(engram.text)
+            deduped.append(engram)
+
+        kept = _filter_by_token_budget(deduped, token_budget)
+
+        return "\n".join(f"- {engram.text}" for engram in kept)
 
     # ---------- rules -------------------
 
@@ -367,8 +402,15 @@ def recall_rules(self) -> str:
         except (OSError, UnicodeDecodeError):
             return ""
 
-    def get_rules(self) -> list[Engram]:
-        """Load behavioral rules (rules.md) as Engrams, grouped by kind."""
+    def get_rules(self, exclude_scratchpad_when: bool = False) -> list[Engram]:
+        """Load behavioral rules (rules.md) as Engrams, grouped by kind.
+
+        exclude_scratchpad_when drops "when" rules that mention "scratchpad"
+        — those are injected separately via recall_scratchpad_wisdom() into
+        the scratchpad tool description; including them in the system prompt
+        too would double their token cost. always/never rules are never
+        affected — those aren't part of that injection.
+        """
         if not self._rules_path.is_file():
             return []
         try:
@@ -390,6 +432,11 @@ def get_rules(self) -> list[Engram]:
 
         entries.sort(key=lambda x: x.kind)
 
+        if exclude_scratchpad_when:
+            entries = [
+                e for e in entries if not (e.kind == "when" and _is_scratchpad_related(e))
+            ]
+
         return entries
 
     def del_rule(self, id):

anton/core/runtime.py

@@ -141,32 +141,29 @@ async def build_chat_session(
     initial_history = history_store.load(session_id)
 
     data_vault = LocalDataVault() if LocalDataVault is not None else None
-    google_drive_oauth_connected = False
     if data_vault is not None:
         try:
-            for conn in data_vault.list_connections():
+            connections = data_vault.list_connections()
+        except Exception:
+            logger.debug("Could not list Anton data vault connections", exc_info=True)
+            connections = []
+        for conn in connections:
+            # Per-connection try/except so one bad record can't abort
+            # injecting env for the rest; engine/name are set before the try
+            # so the except's own logging can't itself raise.
+            engine = None
+            name = None
+            try:
                 engine = conn.get("engine")
                 name = conn.get("name")
                 if not (engine and name):
                     continue
                 data_vault.inject_env(engine, name)
-                if engine == "google_drive":
-                    fields = data_vault.load(engine, name) or {}
-                    if fields.get("auth_type") == "oauth":
-                        google_drive_oauth_connected = True
-        except Exception:
-            logger.debug("Could not inject Anton data vault env", exc_info=True)
-
-    integration_guidance = ""
-    if google_drive_oauth_connected:
-        integration_guidance = (
-            " Connected Google Drive accounts are available through Google OAuth credentials "
-            "in the injected `DS_GOOGLE_DRIVE_<CONNECTION>__...` environment variables. "
-            "Only claim Google Drive access if you can actually use those credentials successfully."
-        )
-
-    suffix_parts = [s for s in (system_prompt_suffix, integration_guidance) if s]
-    final_suffix = "".join(suffix_parts) if suffix_parts else None
+            except Exception:
+                logger.debug(
+                    "Could not inject Anton data vault env for %s/%s",
+                    engine, name, exc_info=True,
+                )
 
     config = ChatSessionConfig(
         llm_client=llm_client,
@@ -176,7 +173,10 @@ async def build_chat_session(
         episodic=episodic,
         system_prompt_context=SystemPromptContext(
             runtime_context=build_runtime_context(settings),
-            suffix=final_suffix,
+            # SystemPromptContext.suffix is typed str (default ""), not
+            # Optional — pass "" rather than None when no suffix was given,
+            # or ChatSystemPromptBuilder.build()'s suffix.strip() crashes.
+            suffix=system_prompt_suffix or "",
         ),
         output_dir=str(output_dir),
         workspace=workspace,