fix(agent): approve teacher-generated skills

src/teacher_escalation.py

@@ -439,56 +439,11 @@ async def escalate_and_learn(
     failure_reason: str,
     owner: Optional[str] = None,
 ) -> Optional[str]:
-    """Call the teacher, evaluate ITS attempt, save a skill on success.
+    """Retire legacy background learning when no approval UI is available."""
-
+    logger.info(
-    Returns the saved skill name (or None if the teacher couldn't
+        "background teacher learning skipped: generated skills require an "
-    write one). Logs but doesn't raise — escalation is best-effort.
+        "interactive exact approval"
-    """
-    from src.settings import get_setting
-    teacher_spec = (get_setting("teacher_model", "") or "").strip()
-    if not teacher_spec:
-        return None
-
-    prompt = _TEACHER_ESCALATION_PROMPT.format(
-        user_request=user_request or "(no user request captured)",
-        failure_reason=failure_reason or "(failure reason not captured)",
-        untrusted_trace_guard=_UNTRUSTED_TRACE_GUARD,
-        trace=_format_trace(tool_results, agent_reply),
     )
-    response = await _call_teacher(teacher_spec, prompt, owner=owner)
-    if not response:
-        return None
-
-    skill = _extract_skill_json(response)
-    if not skill:
-        # Teacher chose not to write a skill — see prompt contract.
-        logger.info("teacher declined to write a skill for this failure")
-        return None
-
-    # Same regex eval applied to the teacher's response — if the
-    # teacher itself sounded uncertain ("I don't have a tool"), drop
-    # the skill rather than persist a sketchy one.
-    status, reason = evaluate_turn_regex([], response)
-    if status == "failure":
-        logger.info(f"teacher response failed eval, skipping skill save: {reason}")
-        return None
-
-    # Tag the skill with the escalation source for auditability.
-    skill.setdefault("source", "teacher-escalation")
-    skill.setdefault("teacher_model", teacher_spec)
-    # Force action=add regardless of what the teacher wrote.
-    skill["action"] = "add"
-
-    import json
-    from src.tool_implementations import do_manage_skills
-    try:
-        result = await do_manage_skills(json.dumps(skill), owner=owner)
-        if isinstance(result, dict) and not result.get("error"):
-            logger.info(f"teacher wrote skill: {skill.get('name')}")
-            return skill.get("name")
-        logger.warning(f"skill save failed: {result}")
-    except Exception as e:
-        logger.warning(f"skill save raised: {e}")
     return None
 
 
@@ -761,31 +716,60 @@ async def run_teacher_inline(
     skill.setdefault("source", "teacher-escalation")
     skill.setdefault("teacher_model", teacher_spec)
 
-    import json as _json
+    if not session_id:
-    from src.tool_implementations import do_manage_skills
-    try:
-        result = await do_manage_skills(_json.dumps(skill), owner=owner)
-        if isinstance(result, dict) and not result.get("error"):
-            logger.info(f"teacher succeeded; saved skill: {skill.get('name')}")
-            yield (
-                'data: ' + json.dumps({
-                    "type": "skill_saved",
-                    "name": skill.get("name"),
-                    "category": skill.get("category", "general"),
-                }) + '\n\n'
-            )
-        else:
-            yield (
-                'data: ' + json.dumps({
-                    "type": "skill_save_failed",
-                    "reason": str(result),
-                }) + '\n\n'
-            )
-    except Exception as e:
-        logger.warning(f"skill save raised: {e}")
         yield (
             'data: ' + json.dumps({
                 "type": "skill_save_failed",
-                "reason": str(e),
+                "reason": (
+                    "Teacher-generated skills require an interactive exact "
+                    "approval before they can be saved."
+                ),
             }) + '\n\n'
         )
+        return
+
+    import json as _json
+    import uuid as _uuid
+    from src.tool_approvals import tool_approval_store
+    from src.tool_capabilities import capabilities_for_action
+
+    skill_content = _json.dumps(skill, ensure_ascii=False)
+    pending = tool_approval_store.create(
+        owner=owner,
+        session_id=session_id,
+        origin_run_id=f"teacher-skill-{_uuid.uuid4().hex}",
+        tool_name="manage_skills",
+        content=skill_content,
+        workspace=workspace,
+        external_untrusted_context_seen=True,
+        capabilities=capabilities_for_action("manage_skills", skill_content),
+    )
+    approval = pending.public_payload(
+        reason=(
+            "The teacher generated this reusable skill. Review and approve "
+            "the complete skill definition before it is saved."
+        ),
+    )
+    yield (
+        "data: "
+        + json.dumps({"delta": "Review the teacher-generated skill before saving it."})
+        + "\n\n"
+    )
+    yield (
+        "data: "
+        + json.dumps({
+            "type": "tool_output",
+            "tool": "manage_skills",
+            "command": str(skill.get("name") or "teacher-generated skill"),
+            "output": "Waiting for an exact user approval.",
+            "exit_code": None,
+            "ask_user": approval,
+            "teacher": True,
+        })
+        + "\n\n"
+    )
+    yield (
+        "data: "
+        + json.dumps({"type": "ask_user", "data": approval, "teacher": True})
+        + "\n\n"
+    )

tests/test_teacher_eval_tier2.py

@@ -171,8 +171,42 @@ async def test_maybe_escalate_tier2_disabled_by_default(monkeypatch):
     assert task is None
 
 
+@pytest.mark.asyncio
+async def test_background_teacher_learning_never_persists_without_approval(monkeypatch):
+    monkeypatch.setattr(
+        "src.settings.get_setting",
+        lambda key, default=None: {
+            "teacher_model": "teacher-model",
+        }.get(key, default),
+    )
+
+    async def fail_teacher_call(*args, **kwargs):
+        raise AssertionError("background learning spent a teacher call without approval UI")
+
+    async def fail_direct_skill_save(*args, **kwargs):
+        raise AssertionError("background teacher output was persisted directly")
+
+    monkeypatch.setattr("src.teacher_escalation._call_teacher", fail_teacher_call)
+    monkeypatch.setattr(
+        "src.tool_implementations.do_manage_skills",
+        fail_direct_skill_save,
+    )
+
+    saved = await teacher_escalation.escalate_and_learn(
+        user_request="test request",
+        tool_results=[],
+        agent_reply="student failed",
+        failure_reason="test failure",
+        owner="alice",
+    )
+
+    assert saved is None
+
+
 @pytest.mark.asyncio
 async def test_run_teacher_inline_triggers_tier2_escalation(monkeypatch):
+    from src.tool_approvals import tool_approval_store
+
     # Settings and gates
     monkeypatch.setattr("src.settings.get_setting", lambda key, default=None: {"teacher_enabled": True, "teacher_model": "teacher-model", "teacher_tier2_enabled": True}.get(key, default))
     monkeypatch.setattr("src.ai_interaction._resolve_model", lambda spec, owner=None: ("http://teacher.local/v1", "teacher-model", {}))
@@ -197,10 +231,13 @@ async def test_run_teacher_inline_triggers_tier2_escalation(monkeypatch):
         return '```json\n{"action": "add", "name": "test-skill"}\n```'
     monkeypatch.setattr("src.teacher_escalation._call_teacher", fake_call_teacher)
 
-    # Mock do_manage_skills
+    async def fail_direct_skill_save(*args, **kwargs):
-    async def fake_do_manage_skills(skill_json, owner=None):
+        raise AssertionError("teacher output was persisted without approval")
-        return {"success": True}
+
-    monkeypatch.setattr("src.tool_implementations.do_manage_skills", fake_do_manage_skills)
+    monkeypatch.setattr(
+        "src.tool_implementations.do_manage_skills",
+        fail_direct_skill_save,
+    )
 
     events = []
     async for evt in teacher_escalation.run_teacher_inline(
@@ -209,13 +246,34 @@ async def test_run_teacher_inline_triggers_tier2_escalation(monkeypatch):
         student_tool_events=[],
         student_reply="student reply",
         owner="alice",
+        session_id="teacher-approval-session",
     ):
         events.append(evt)
 
-    # Make sure teacher takeover was announced and executed
+    # The teacher takeover runs, but its cross-model skill output is sealed for
+    # an explicit approval instead of being written directly.
     assert any("teacher_takeover" in evt for evt in events)
     assert any("tool_output" in evt for evt in events)
-    assert any("skill_saved" in evt for evt in events)
+    approval_event = next(
+        json.loads(evt[6:])
+        for evt in events
+        if evt.startswith("data: ")
+        and "\"kind\": \"tool_approval\"" in evt
+        and "\"type\": \"tool_output\"" in evt
+    )
+    approval = approval_event["ask_user"]
+    pending = tool_approval_store.peek(approval["approval_id"])
+    assert pending is not None
+    assert pending.tool_name == "manage_skills"
+    assert json.loads(pending.content)["name"] == "test-skill"
+    assert pending.external_untrusted_context_seen is True
+    tool_approval_store.consume(
+        pending.approval_id,
+        decision="deny",
+        owner="alice",
+        session_id="teacher-approval-session",
+    )
+    assert not any("skill_saved" in evt for evt in events)
 
 
 @pytest.mark.asyncio