[pierre] fix: keep the App Builder's runtime contract current on every install

This commit is contained in:
SirKentut
2026-07-10 17:54:49 -07:00
parent f542007a3c
commit b40cb79216
4 changed files with 172 additions and 16 deletions
@@ -12,6 +12,7 @@ from typeguard import typechecked
from backend.apps.agents.core.models import AgentSession
from backend.apps.agents.manager.prompt.tool_catalog import get_all_tool_names
from backend.apps.agents.manager.prompt.prompt_context import (
build_app_runtime_contract,
build_browser_context,
build_installed_skills_catalog,
build_mcp_registry_summary,
@@ -73,6 +74,9 @@ def compose_turn_system_prompt(
from backend.apps.outputs.view_builder_templates import load_app_builder_skill
skill_block = f"<app_builder_reference>\n{load_app_builder_skill()}\n</app_builder_reference>"
composed_prompt = f"{composed_prompt}\n\n{skill_block}" if composed_prompt else skill_block
# Appended AFTER the reference, and never sourced from it: the skill is a user-editable file seeded once per install, so a stale or edited copy silently drops whatever it omits. Platform mechanics have to reach the agent regardless of what that file says.
contract_block = build_app_runtime_contract(session.cwd)
composed_prompt = f"{composed_prompt}\n\n{contract_block}"
else:
# Every other mode gets one line of discovery instead of the whole reference: CreateApp's result carries the reference when actually used, so the base prompt stays cheap.
apps_note = (
@@ -1,3 +1,4 @@
import os
from typing import Callable, Dict, List, Optional, Tuple
from typeguard import typechecked
@@ -196,6 +197,37 @@ def build_selected_app_context(selected_app_output_ids: Optional[List[str]]) ->
)
@typechecked
def build_app_runtime_contract(workspace_path: Optional[str]) -> str:
"""The non-negotiable runtime mechanics for an App Builder turn: where the app's
terminal lives, how to restart it, and the requirement to read it before claiming
a change works. Deliberately NOT sourced from the App Builder skill file, which is
seeded once per install and never overwritten, so an install that predates a skill
update (or a user who edits the guidance out) would otherwise never see any of this."""
root = workspace_path or "."
log = os.path.join(root, ".openswarm", "terminal.log")
return (
"<app_runtime_contract>\n"
"Your app is already running. Its terminal — backend stdout/stderr, runtime events, and the\n"
"browser console — is tee'd to a file you can read directly. This is the ONLY way you can see\n"
"what the app actually does; editing files tells you nothing about whether it runs.\n\n"
"Read it with exactly this, every time:\n\n"
f' tail -50 {log} 2>/dev/null || echo "Terminal log not yet available"\n\n'
"Lines are prefixed [BACKEND], [BACKEND:stderr], [RUNTIME], [FRONTEND], [FRONTEND:warn],\n"
"[FRONTEND:error]. Grep it for `error` when it is long. If this app is open in more than one\n"
"dashboard card, the extra cards log to terminal-2.log, terminal-3.log, and so on.\n\n"
"Rules, not suggestions:\n"
"- Read the terminal after every batch of writes. Fix what it reports before moving on.\n"
f"- Read it again before you tell the user anything is done. `bash {os.path.join(root, 'restart.sh')}` restarts the\n"
" runtime; do that after installing packages or changing backend startup, then read the log to\n"
" confirm a clean boot. The file resets on every start.\n"
"- \"Terminal log not yet available\" means the runtime never started. That is a problem to fix,\n"
" not a reason to skip the check.\n"
"- Never claim the app works when you have not read the terminal. If you did not check, say so.\n"
"</app_runtime_contract>"
)
@typechecked
def build_selected_settings_context(selected_setting_ids: Optional[List[str]]) -> Optional[str]:
"""Context block when the user points the agent at specific Settings rows.
+46 -16
View File
@@ -1,4 +1,5 @@
import os
import hashlib
import json
import logging
import re
@@ -106,34 +107,63 @@ def p_built_in_skill_registry() -> list[dict]:
]
def p_seed_built_in_skills() -> None:
"""Copy each built-in skill into SKILLS_DIR if not already present, and
ensure the index has the `built_in: true` flag so the UI and DELETE
endpoint know to treat it specially. Idempotent; safe to call on
every boot. Doesn't overwrite the file once it exists (so user edits
are preserved across restarts and upgrades)."""
def p_content_hash(text: str) -> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest()
def seed_built_in_skills() -> None:
"""Copy each built-in skill into SKILLS_DIR and keep an *unedited* copy in sync
with the bundled source across upgrades. Idempotent; safe on every boot.
`seeded_hash` in the index records the bytes we last wrote. A file still hashing
to it was never edited, so a newer bundle replaces it; a file that diverges is a
user edit and is left alone. Installs predating `seeded_hash` can't be told apart
from an edit, so we only claim provenance when the bytes already match the bundle
-- otherwise they stay untracked and frozen, since silently clobbering a real edit
is the worse failure. Before this, seeding was create-if-absent, so every install
was pinned forever to whatever shipped the day it first booted."""
index = load_index()
dirty = False
for entry in p_built_in_skill_registry():
skill_id = entry["id"]
fpath = os.path.join(SKILLS_DIR, f"{skill_id}.md")
if not os.path.exists(fpath):
try:
with open(entry["source_path"], encoding="utf-8") as src:
bundled = src.read()
except OSError:
logger.warning("built-in skill source missing: %s", entry["source_path"])
continue
current = None
if os.path.exists(fpath):
try:
with open(entry["source_path"], encoding="utf-8") as src:
content = src.read()
with open(fpath, "w", encoding="utf-8") as dst:
dst.write(content)
except FileNotFoundError:
logger.warning("built-in skill source missing: %s", entry["source_path"])
with open(fpath, encoding="utf-8") as f:
current = f.read()
except OSError:
logger.warning("built-in skill unreadable, leaving as-is: %s", fpath, exc_info=True)
continue
# Refresh index metadata. Existing user-changed name/description in the index stays, but built_in always gets re-asserted in case the index was created before this mechanism existed.
meta = dict(index.get(skill_id, {}))
seeded_hash = meta.get("seeded_hash")
if current is None or (seeded_hash and p_content_hash(current) == seeded_hash):
# Absent, or byte-identical to what we last seeded: no user edit to lose.
if current != bundled:
try:
os.makedirs(SKILLS_DIR, exist_ok=True)
with open(fpath, "w", encoding="utf-8") as dst:
dst.write(bundled)
except OSError:
logger.warning("built-in skill write failed: %s", fpath, exc_info=True)
continue
meta["seeded_hash"] = p_content_hash(bundled)
elif not seeded_hash and current == bundled:
# Untracked but already in sync; safe to adopt so the NEXT upgrade can move it.
meta["seeded_hash"] = p_content_hash(bundled)
# Anything else is a user edit, or an untracked install indistinguishable from one: leave both the file and its (absent) provenance alone so we never overwrite it.
# Refresh index metadata. Existing user-changed name/description in the index stays, but built_in always gets re-asserted in case the index was created before this mechanism existed.
meta.setdefault("name", entry["name"])
meta.setdefault("description", entry["description"])
meta.setdefault("command", entry["command"])
if not meta.get("built_in"):
meta["built_in"] = True
dirty = True
if index.get(skill_id) != meta:
index[skill_id] = meta
dirty = True
@@ -156,7 +186,7 @@ async def skills_lifespan():
os.makedirs(SKILLS_DIR, exist_ok=True)
os.makedirs(SKILLS_WORKSPACE_DIR, exist_ok=True)
try:
p_seed_built_in_skills()
seed_built_in_skills()
p_prune_orphan_index()
except Exception:
# Don't block app startup on a skill-seed failure; the worst case is the user has to manually paste the skill in once.
@@ -0,0 +1,90 @@
"""Built-in skills must follow the bundled source across upgrades unless the user edited them.
Seeding used to be create-if-absent, which pinned every install to whatever shipped the day it
first booted: the App Builder agent's prompt kept a months-old skill and so never learned that
`.openswarm/terminal.log` existed. `seeded_hash` records the bytes we last wrote so an untouched
file can be safely replaced, while a real edit (or an untracked pre-existing install) is left alone.
"""
from __future__ import annotations
import json
import os
import pytest
import backend.apps.skills.skills as skills_mod
@pytest.fixture
def seeded(tmp_path, monkeypatch):
"""A stubbed SKILLS_DIR plus a one-entry built-in registry pointing at a bundle we control."""
d = tmp_path / "skills"
d.mkdir()
monkeypatch.setattr(skills_mod, "SKILLS_DIR", str(d))
monkeypatch.setattr(skills_mod, "INDEX_PATH", str(d / ".skills_index.json"))
source = tmp_path / "bundled.md"
source.write_text("v1 bundled", encoding="utf-8")
def p_registry():
return [{
"id": "app_builder_skill",
"name": "App Builder",
"description": "d",
"command": "app-builder-skill",
"source_path": str(source),
}]
# String-literal setattr: p_built_in_skill_registry stays file-private under the p-private rule.
monkeypatch.setattr(skills_mod, "p_built_in_skill_registry", p_registry)
return d, source
def p_skill(d):
return os.path.join(str(d), "app_builder_skill.md")
def p_read(d):
return open(p_skill(d), encoding="utf-8").read()
def test_unedited_copy_upgrades_when_the_bundle_changes(seeded):
d, source = seeded
skills_mod.seed_built_in_skills()
source.write_text("v2 bundled with terminal.log", encoding="utf-8")
skills_mod.seed_built_in_skills()
assert p_read(d) == "v2 bundled with terminal.log"
def test_user_edit_is_preserved_across_a_bundle_bump(seeded):
d, source = seeded
skills_mod.seed_built_in_skills()
with open(p_skill(d), "w", encoding="utf-8") as f:
f.write("my own house rules")
source.write_text("v2 bundled", encoding="utf-8")
skills_mod.seed_built_in_skills()
assert p_read(d) == "my own house rules"
def test_second_boot_does_not_clobber_a_preserved_edit(seeded):
# Regression: adopting the *current* bytes as provenance would make the next boot treat a real edit as unedited.
d, source = seeded
skills_mod.seed_built_in_skills()
with open(p_skill(d), "w", encoding="utf-8") as f:
f.write("my own house rules")
source.write_text("v2 bundled", encoding="utf-8")
skills_mod.seed_built_in_skills()
skills_mod.seed_built_in_skills()
assert p_read(d) == "my own house rules"
def test_untracked_stale_install_is_never_clobbered(seeded):
# The real-world bug: a file seeded before seeded_hash existed. Indistinguishable from an edit, so leave it.
d, source = seeded
with open(p_skill(d), "w", encoding="utf-8") as f:
f.write("stale bundle from an old install")
source.write_text("v2 bundled", encoding="utf-8")
skills_mod.seed_built_in_skills()
assert p_read(d) == "stale bundle from an old install"
with open(d / ".skills_index.json", encoding="utf-8") as f:
assert "seeded_hash" not in json.load(f)["app_builder_skill"]