skill and develop

2026-08-27 14:18:51 +02:00
parent 8119f780e7
commit f78f667823
4 changed files with 682 additions and 70 deletions
--- a/skills/compact-memory/scripts/compact_memory_auto.py
+++ b/skills/compact-memory/scripts/compact_memory_auto.py
@@ -3,49 +3,122 @@
 # requires-python = ">=3.11"
 # dependencies = ["nanobot-ai"]
 # ///
-"""Nightly compact-memory auto run, triggered by the nanobot user crontab.
+"""Nightly compact-memory run, triggered by the nanobot user crontab.

-Runs the compact-memory skill in auto mode via the Nanobot Python API with a
-fresh, never-reused `session_key` per invocation, then delivers the result
-directly to Telegram. This bypasses nanobot's cron/jobs.json `agent_turn`
-mechanism entirely, so the turn never gets appended to the user's live chat
-session (which caused context contamination across nights) and needs no
-delivery-gating evaluator (there is none for bound cron jobs in nanobot-ai
-0.2.2 — whatever the agent answers would otherwise go straight to the chat).
+The agent only decides *what* to remove or merge and returns it as a JSON change-set;
+this script validates it, applies it, writes the backup and the audit log, and composes
+the Telegram message. Nothing the agent writes as prose ever reaches Telegram, so the
+delivered report no longer depends on the model following output-format instructions.

-Same pattern as skills/remind/scripts/remind_send.py and
-skills/detach/scripts/tasks-daemon.py: external script, direct Telegram Bot
-API delivery, no nanobot channel pipeline involved.
+Runs with a fresh, never-reused `session_key` per invocation and delivers straight to the
+Telegram Bot API, bypassing nanobot's cron/jobs.json `agent_turn` mechanism (which would
+append the turn to the user's live chat session and deliver whatever the agent answered).
+
+Same external-script pattern as skills/remind/scripts/remind_send.py and
+skills/detach/scripts/tasks-daemon.py.
 """

 from __future__ import annotations

 import asyncio
+import hashlib
 import json
+import re
 import sys
 import traceback
 import urllib.parse
 import urllib.request
+from dataclasses import dataclass
 from datetime import datetime
 from pathlib import Path
+from typing import TYPE_CHECKING, Any

-from nanobot import Nanobot
+if TYPE_CHECKING:
+    from nanobot import Nanobot  # ty: ignore[unresolved-import]

 CONFIG = Path.home() / ".nanobot" / "config.json"
+WORKSPACE_FALLBACK = Path.home() / ".nanobot" / "workspace"
+MEMORY_REL = "memory/MEMORY.md"
+BACKUP_REL = "backup"
+CLEAN_LOG_REL = "log/memory-clean.log"
+
 FALLBACK_CHAT_ID = "8826147089"
 TIMEOUT_SECONDS = 10 * 60
+MAX_ATTEMPTS = 3
+MODEL_PRESET = "kimi27"

-GOAL = (
-    "Read skills/compact-memory/SKILL.md and run its Auto mode to audit and "
-    "compact memory/MEMORY.md. Perform every step yourself using your file "
-    "tools — there is no script to run."
-)
+DELETE_CATEGORIES = frozenset({"superseded", "detail", "duplicate", "ephemeral", "stale-section"})
+DELETE_KEYS = frozenset({"op", "category", "original", "reason"})
+MERGE_KEYS = frozenset({"op", "original", "new_text", "reason"})
+
+MAX_REASON_CHARS = 120
+MAX_NEW_TEXT_LINES = 3
+MAX_NEW_TEXT_CHARS = 300
+MAX_ORIGINAL_LINES = 20
+MAX_REMOVED_RATIO = 0.5
+QUOTE_CHARS = 100
+MAX_ERROR_CHARS = 300
+
+JSON_FENCE = re.compile(r"```(?:json)?\s*\n(.*?)```", re.DOTALL)
+
+GOAL = """\
+Read skills/compact-memory/SKILL.md and run its Nightly mode to audit memory/MEMORY.md.
+
+This is an unattended nightly run. Do NOT edit, write or create any file — not memory/MEMORY.md,
+not a backup, not a log entry. Your only job is to decide what should be removed or merged and
+to hand it over as a change-set; a script applies it.
+
+Answer with exactly one ```json code block following the Nightly mode schema in the skill, and
+nothing else — no narration, no summary, no text before or after the block. If nothing qualifies,
+answer with a block whose "changes" list is empty.
+"""
+
+RETRY_PROMPT = """\
+Your change-set was rejected by the validator:
+
+{errors}
+
+Answer again with exactly one corrected ```json code block and nothing else. Copy every `original`
+line character-for-character from memory/MEMORY.md — read the file again if you are unsure.
+"""


-def _telegram_config() -> tuple[str, str]:
+class CompactMemoryError(Exception):
+    """A failure worth reporting to the user in one line."""
+
+
+class ChangeSetError(CompactMemoryError):
+    """The agent's change-set is malformed or does not match the current file."""
+
+
+@dataclass(frozen=True)
+class Change:
+    op: str
+    category: str
+    original: tuple[str, ...]
+    new_text: tuple[str, ...]
+    reason: str
+
+
+@dataclass(frozen=True)
+class LocatedChange:
+    change: Change
+    start: int
+    end: int
+
+
+def _config() -> dict[str, Any]:
+    return json.loads(CONFIG.read_text(encoding="utf-8"))
+
+
+def _workspace(config: dict[str, Any]) -> Path:
+    configured = config.get("agents", {}).get("defaults", {}).get("workspace")
+    return Path(configured).expanduser() if configured else WORKSPACE_FALLBACK
+
+
+def _telegram_config(config: dict[str, Any]) -> tuple[str, str]:
    """Return (bot token, chat id). Chat id reads channels.telegram.allowFrom[0], with a constant fallback."""
-    data = json.loads(CONFIG.read_text(encoding="utf-8"))
-    telegram = data["channels"]["telegram"]
+    telegram = config["channels"]["telegram"]
    allow_from = telegram.get("allowFrom") or []
    chat_id = str(allow_from[0]) if allow_from else FALLBACK_CHAT_ID
    return telegram["token"], chat_id
@@ -59,36 +132,269 @@ def _send_telegram(text: str, token: str, chat_id: str) -> None:
        resp.read()


-async def _run_audit(session_key: str) -> str:
-    bot = Nanobot.from_config()
-    result = await bot.run(GOAL, session_key=session_key)
-    return result.content or ""
+def _sha256(path: Path) -> str:
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def _shorten(text: str, limit: int) -> str:
+    return text if len(text) <= limit else text[: limit - 1].rstrip() + "…"
+
+
+def _quote(lines: tuple[str, ...]) -> str:
+    return _shorten(" ".join(line.strip() for line in lines), QUOTE_CHARS)
+
+
+def _extract_json(content: str) -> Any:
+    """Return the JSON payload from the last parseable fenced block, or from the whole answer."""
+    candidates = [*JSON_FENCE.findall(content), content]
+    for candidate in reversed(candidates):
+        try:
+            return json.loads(candidate.strip())
+        except json.JSONDecodeError:
+            continue
+    raise ChangeSetError("- the answer contains no parseable ```json block")
+
+
+def _string_list(raw: Any, field: str, index: int) -> tuple[str, ...]:
+    if not isinstance(raw, list) or not raw or not all(isinstance(line, str) for line in raw):
+        raise ChangeSetError(f'- item {index}: "{field}" must be a non-empty list of strings')
+    return tuple(raw)
+
+
+def _parse_change(raw: Any, index: int) -> Change:
+    if not isinstance(raw, dict):
+        raise ChangeSetError(f"- item {index}: must be a JSON object")
+
+    op = raw.get("op")
+    if op not in ("delete", "merge"):
+        raise ChangeSetError(f'- item {index}: "op" must be "delete" or "merge"')
+
+    allowed = DELETE_KEYS if op == "delete" else MERGE_KEYS
+    unknown = sorted(str(key) for key in set(raw) - allowed)
+    if unknown:
+        raise ChangeSetError(f"- item {index}: unknown fields {unknown}, allowed are {sorted(allowed)}")
+
+    reason = raw.get("reason")
+    if not isinstance(reason, str) or not reason.strip():
+        raise ChangeSetError(f'- item {index}: "reason" must be a non-empty string')
+    if len(reason) > MAX_REASON_CHARS:
+        raise ChangeSetError(f'- item {index}: "reason" has {len(reason)} characters, the limit is {MAX_REASON_CHARS}')
+
+    original = _string_list(raw.get("original"), "original", index)
+    if len(original) > MAX_ORIGINAL_LINES:
+        raise ChangeSetError(f'- item {index}: "original" has {len(original)} lines, the limit is {MAX_ORIGINAL_LINES}')
+
+    if op == "delete":
+        category = raw.get("category")
+        if category not in DELETE_CATEGORIES:
+            raise ChangeSetError(f'- item {index}: "category" must be one of {sorted(DELETE_CATEGORIES)}')
+        return Change(op=op, category=category, original=original, new_text=(), reason=reason.strip())
+
+    new_text = _string_list(raw.get("new_text"), "new_text", index)
+    if len(new_text) > MAX_NEW_TEXT_LINES:
+        raise ChangeSetError(f'- item {index}: "new_text" has {len(new_text)} lines, the limit is {MAX_NEW_TEXT_LINES}')
+    joined = "\n".join(new_text)
+    if len(joined) > MAX_NEW_TEXT_CHARS:
+        raise ChangeSetError(
+            f'- item {index}: "new_text" has {len(joined)} characters, the limit is {MAX_NEW_TEXT_CHARS}'
+        )
+    if len(new_text) >= len(original):
+        raise ChangeSetError(f'- item {index}: a merge must produce fewer lines than "original" has')
+    return Change(op=op, category="merge", original=original, new_text=new_text, reason=reason.strip())
+
+
+def _find_block(lines: list[str], block: tuple[str, ...]) -> list[int]:
+    """Return every start index where the block matches whole lines (ignoring trailing whitespace)."""
+    needle = [line.rstrip() for line in block]
+    haystack = [line.rstrip() for line in lines]
+    span = len(needle)
+    return [start for start in range(len(haystack) - span + 1) if haystack[start : start + span] == needle]
+
+
+def _locate(changes: list[Change], lines: list[str]) -> list[LocatedChange]:
+    errors: list[str] = []
+    located: list[LocatedChange] = []
+    for index, change in enumerate(changes, start=1):
+        matches = _find_block(lines, change.original)
+        if not matches:
+            errors.append(f'- item {index}: "original" does not appear in MEMORY.md: {_quote(change.original)}')
+        elif len(matches) > 1:
+            errors.append(f'- item {index}: "original" appears {len(matches)} times, it must be unique')
+        else:
+            located.append(LocatedChange(change=change, start=matches[0], end=matches[0] + len(change.original)))
+    if errors:
+        raise ChangeSetError("\n".join(errors))
+    return sorted(located, key=lambda item: item.start)
+
+
+def _check_spans(located: list[LocatedChange], total_lines: int) -> None:
+    cursor = 0
+    for item in located:
+        if item.start < cursor:
+            raise ChangeSetError(f"- overlapping items around line {item.start + 1}, each block must be separate")
+        cursor = item.end
+    removed = sum(len(item.change.original) - len(item.change.new_text) for item in located)
+    if removed > total_lines * MAX_REMOVED_RATIO:
+        raise ChangeSetError(
+            f"- the change-set removes {removed} of {total_lines} lines, more than "
+            f"{int(MAX_REMOVED_RATIO * 100)}% of the file"
+        )
+
+
+def parse_change_set(content: str, memory_text: str) -> list[LocatedChange]:
+    """Validate the agent's answer against the current file. Raises ChangeSetError on any problem."""
+    payload = _extract_json(content)
+    if not isinstance(payload, dict):
+        raise ChangeSetError("- the JSON block must be an object")
+    raw_changes = payload.get("changes")
+    if not isinstance(raw_changes, list):
+        raise ChangeSetError('- the JSON object must have a "changes" list')
+
+    errors: list[str] = []
+    changes: list[Change] = []
+    for index, raw in enumerate(raw_changes, start=1):
+        try:
+            changes.append(_parse_change(raw, index))
+        except ChangeSetError as error:
+            errors.append(str(error))
+    if errors:
+        raise ChangeSetError("\n".join(errors))
+
+    lines = memory_text.splitlines()
+    located = _locate(changes, lines)
+    _check_spans(located, len(lines))
+    return located
+
+
+def _rebuild(lines: list[str], located: list[LocatedChange]) -> list[str]:
+    result: list[str] = []
+    cursor = 0
+    for item in located:
+        result.extend(lines[cursor : item.start])
+        result.extend(item.change.new_text)
+        cursor = item.end
+    result.extend(lines[cursor:])
+    return result
+
+
+def _log_line(change: Change, stamp: str) -> str:
+    quoted = _quote(change.original)
+    if change.op == "delete":
+        return f'{stamp} DELETED [{change.category}] "{quoted}" — {change.reason}'
+    return f'{stamp} MERGED [merge] "{quoted}" → "{_quote(change.new_text)}" — {change.reason}'
+
+
+def apply_change_set(memory: Path, located: list[LocatedChange], workspace: Path, now: datetime) -> int:
+    """Back up the file, apply the change-set, append the audit log. Returns the new line count."""
+    text = memory.read_text(encoding="utf-8")
+    backup_dir = workspace / BACKUP_REL
+    backup_dir.mkdir(parents=True, exist_ok=True)
+    (backup_dir / f"{now:%Y-%m-%d_%H%M}_memory.backup.md").write_text(text, encoding="utf-8")
+
+    new_lines = _rebuild(text.splitlines(), located)
+    memory.write_text("\n".join(new_lines) + "\n", encoding="utf-8")
+
+    clean_log = workspace / CLEAN_LOG_REL
+    clean_log.parent.mkdir(parents=True, exist_ok=True)
+    stamp = f"{now:%Y-%m-%d %H:%M}"
+    with clean_log.open("a", encoding="utf-8") as handle:
+        for item in located:
+            handle.write(_log_line(item.change, stamp) + "\n")
+
+    return len(new_lines)
+
+
+def format_report(located: list[LocatedChange], lines_before: int, lines_after: int) -> str:
+    if not located:
+        return f"Memory compact: nothing to remove (MEMORY.md, {lines_before} lines)."
+
+    deleted = sum(1 for item in located if item.change.op == "delete")
+    merged = len(located) - deleted
+    rows = [f"Memory compact: deleted {deleted}, merged {merged} ({lines_before} → {lines_after} lines)."]
+    for item in located:
+        change = item.change
+        quoted = _quote(change.original)
+        if change.op == "delete":
+            rows.append(f'- [{change.category}] "{quoted}" — {change.reason}')
+        else:
+            rows.append(f'- [merge] "{quoted}" → "{_quote(change.new_text)}" — {change.reason}')
+    return "\n".join(rows)
+
+
+async def _resolve_change_set(bot: Nanobot, session_key: str, memory: Path) -> list[LocatedChange]:
+    """Ask the agent for a change-set, retrying with validator feedback in the same session."""
+    memory_text = memory.read_text(encoding="utf-8")
+    memory_hash = _sha256(memory)
+    message = GOAL
+    last_error = ChangeSetError("- no attempt was made")
+
+    for attempt in range(1, MAX_ATTEMPTS + 1):
+        result = await bot.run(message, session_key=session_key)
+        content = result.content or ""
+        print(f"[attempt {attempt}/{MAX_ATTEMPTS}] answer:\n{content}", file=sys.stderr)
+
+        if _sha256(memory) != memory_hash:
+            raise CompactMemoryError("the agent changed MEMORY.md itself although it must not — nothing applied")
+
+        try:
+            return parse_change_set(content, memory_text)
+        except ChangeSetError as error:
+            last_error = error
+            print(f"[attempt {attempt}/{MAX_ATTEMPTS}] rejected:\n{error}", file=sys.stderr)
+            message = RETRY_PROMPT.format(errors=error)
+
+    first_reason = str(last_error).splitlines()[0].lstrip("- ")
+    raise CompactMemoryError(f"after {MAX_ATTEMPTS} attempts: {first_reason}")
+
+
+async def _run(session_key: str, workspace: Path, now: datetime) -> str:
+    # Deferred so the change-set logic can be imported and tested without nanobot-ai installed.
+    from nanobot import Nanobot  # ty: ignore[unresolved-import]
+
+    memory = workspace / MEMORY_REL
+    lines_before = len(memory.read_text(encoding="utf-8").splitlines())
+
+    bot = Nanobot.from_config(model_preset=MODEL_PRESET)
+    located = await _resolve_change_set(bot, session_key, memory)
+    if not located:
+        return format_report([], lines_before, lines_before)
+
+    lines_after = apply_change_set(memory, located, workspace, now)
+    return format_report(located, lines_before, lines_after)


 def main() -> int:
-    session_key = f"compact-memory-auto:{datetime.now():%Y%m%d-%H%M%S}"
+    now = datetime.now()
+    session_key = f"compact-memory-auto:{now:%Y%m%d-%H%M%S}"
+    config = _config()
+    workspace = _workspace(config)

    try:
-        content = asyncio.run(asyncio.wait_for(_run_audit(session_key), timeout=TIMEOUT_SECONDS))
-    except asyncio.TimeoutError:
-        print(f"compact_memory_auto: timeout after {TIMEOUT_SECONDS // 60} min (session={session_key})", file=sys.stderr)
-        return 1
-    except Exception as e:
-        print(f"compact_memory_auto: run failed (session={session_key}): {e}\n{traceback.format_exc()}", file=sys.stderr)
-        return 1
+        message = asyncio.run(asyncio.wait_for(_run(session_key, workspace, now), timeout=TIMEOUT_SECONDS))
+        failed = False
+    except TimeoutError:
+        message = f"Memory compact: ERROR — timed out after {TIMEOUT_SECONDS // 60} min."
+        failed = True
+    except CompactMemoryError as error:
+        message = _shorten(f"Memory compact: ERROR — {error}.", MAX_ERROR_CHARS)
+        failed = True
+    except Exception as error:  # noqa: BLE001 — the nightly job must report, not just die
+        traceback.print_exc()
+        message = _shorten(f"Memory compact: ERROR — {type(error).__name__}: {error}.", MAX_ERROR_CHARS)
+        failed = True

-    if not content.strip():
-        print(f"compact_memory_auto: empty response (session={session_key})", file=sys.stderr)
-        return 1
+    if failed:
+        message += " Nothing applied, details in log/compact_memory_auto_cron.log."
+        print(f"compact_memory_auto: {message} (session={session_key})", file=sys.stderr)

-    token, chat_id = _telegram_config()
+    token, chat_id = _telegram_config(config)
    try:
-        _send_telegram(content, token, chat_id)
-    except Exception as e:
-        print(f"compact_memory_auto: telegram delivery failed (session={session_key}): {e}", file=sys.stderr)
+        _send_telegram(message, token, chat_id)
+    except Exception as error:  # noqa: BLE001 — delivery failure must not hide the original outcome
+        print(f"compact_memory_auto: telegram delivery failed (session={session_key}): {error}", file=sys.stderr)
        return 1

-    return 0
+    return 1 if failed else 0


 if __name__ == "__main__":