feat(skill-evolution): add SkillReviewWorker for passive post-turn skill creation

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

offline-backend-20260512/backend/packages/harness/deerflow/skills/review_worker.py

@@ -0,0 +1,183 @@
+"""Post-turn background skill review worker.
+
+After a conversation turn completes, SkillReviewMiddleware calls
+``maybe_start_review()``.  This module runs a lightweight LLM React agent
+(with skill_manage + skill_list tools) that reads the conversation history
+and decides whether to create or patch skills.
+
+Concurrency model: per-thread asyncio.Lock + a single _pending slot.
+If a review is already running for the thread, the newer messages overwrite
+the pending slot; once the current review finishes, it drains that slot.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import textwrap
+from typing import TYPE_CHECKING, Any
+
+from langchain_core.messages import BaseMessage
+
+if TYPE_CHECKING:
+    from deerflow.config.app_config import AppConfig
+
+logger = logging.getLogger(__name__)
+
+_locks: dict[str, asyncio.Lock] = {}
+_pending: dict[str, list[BaseMessage]] = {}
+
+
+def _get_lock(thread_id: str) -> asyncio.Lock:
+    if thread_id not in _locks:
+        _locks[thread_id] = asyncio.Lock()
+    return _locks[thread_id]
+
+
+def extract_tool_call_count(messages: list[BaseMessage]) -> int:
+    """Count total tool calls across all AI messages in the conversation."""
+    count = 0
+    for msg in messages:
+        tool_calls: Any = None
+        if hasattr(msg, "tool_calls"):
+            tool_calls = msg.tool_calls
+        elif isinstance(msg, dict):
+            tool_calls = msg.get("tool_calls")
+        if tool_calls:
+            count += len(tool_calls)
+    return count
+
+
+def _format_messages_for_review(messages: list[BaseMessage], max_messages: int) -> str:
+    """Serialize recent messages to a readable text block for the LLM reviewer."""
+    recent = messages[-max_messages:]
+    lines: list[str] = []
+    for msg in recent:
+        role = type(msg).__name__.replace("Message", "")
+        content = msg.content if isinstance(msg.content, str) else str(msg.content)
+        tool_calls = getattr(msg, "tool_calls", None)
+        if tool_calls:
+            calls_str = ", ".join(
+                f"{tc.name if hasattr(tc, 'name') else tc.get('name', '?')}()"
+                for tc in tool_calls
+            )
+            lines.append(f"[{role}] <tool_calls: {calls_str}>")
+        else:
+            preview = content[:400].replace("\n", " ")
+            lines.append(f"[{role}] {preview}")
+    return "\n".join(lines)
+
+
+_REVIEW_SYSTEM_PROMPT = textwrap.dedent("""\
+    You are a skill review agent. Your job is to analyze a completed conversation
+    and decide whether any reusable multi-step workflow should be saved as a skill.
+
+    ## Decision rules
+    1. Call skill_list first to see existing skills — avoid creating duplicates.
+    2. CREATE a new skill if: the conversation demonstrates a non-trivial, repeatable
+       workflow (3+ distinct tool operations) that a future agent could reuse verbatim.
+    3. PATCH an existing skill if: the conversation revealed a gap, error, or improvement
+       to a skill that was already invoked.
+    4. DO NOTHING if: the task was one-off, data-specific, trivial, or already covered.
+
+    ## Skill quality bar
+    - Name: kebab-case, describes WHAT the skill does (e.g. "git-bisect-bug-hunt").
+    - Description: one sentence starting with "When" that states the trigger condition.
+    - Content: concise workflow guide (steps, commands, pitfalls) — NOT a transcript.
+    - Never embed user-specific data, file paths, or credentials.
+
+    Be conservative. Fewer high-quality skills beat many noisy ones.
+""")
+
+
+async def _run_review(thread_id: str, messages: list[BaseMessage], app_config: "AppConfig") -> None:
+    """Core review logic: spin up a minimal React agent and invoke it."""
+    from langgraph.prebuilt import create_react_agent
+
+    from deerflow.models.factory import create_chat_model
+    from deerflow.tools.builtins.skill_tools import skill_list_tool
+    from deerflow.tools.skill_manage_tool import skill_manage_tool
+
+    evolution_cfg = app_config.skill_evolution
+    timeout = evolution_cfg.review_worker.timeout_seconds
+    max_messages = evolution_cfg.review_worker.max_messages
+    model_name = evolution_cfg.review_worker.model_name
+
+    conversation_text = _format_messages_for_review(messages, max_messages)
+
+    try:
+        model = create_chat_model(model_name)
+    except Exception:
+        logger.warning("skill_review_worker: could not load model %r, aborting review", model_name)
+        return
+
+    tools = [skill_list_tool, skill_manage_tool]
+    agent = create_react_agent(model, tools, prompt=_REVIEW_SYSTEM_PROMPT)
+
+    user_message = (
+        "Review this conversation and apply any skill changes you deem appropriate.\n\n"
+        f"<conversation>\n{conversation_text}\n</conversation>"
+    )
+
+    try:
+        await asyncio.wait_for(
+            asyncio.get_event_loop().run_in_executor(
+                None,
+                lambda: agent.invoke({"messages": [("user", user_message)]}),
+            ),
+            timeout=timeout,
+        )
+        logger.info("skill_review_worker: review completed for thread %s", thread_id)
+    except TimeoutError:
+        logger.warning("skill_review_worker: review timed out after %ds (thread %s)", timeout, thread_id)
+    except Exception:
+        logger.exception("skill_review_worker: review failed for thread %s", thread_id)
+
+
+async def _run_review_with_pending(
+    thread_id: str,
+    messages: list[BaseMessage],
+    app_config: "AppConfig",
+) -> None:
+    """Run one review; if a newer snapshot arrived while we ran, drain it."""
+    lock = _get_lock(thread_id)
+    async with lock:
+        await _run_review(thread_id, messages, app_config)
+
+    pending = _pending.pop(thread_id, None)
+    if pending is not None:
+        asyncio.create_task(_run_review_with_pending(thread_id, pending, app_config))
+
+
+def maybe_start_review(
+    thread_id: str,
+    messages: list[BaseMessage],
+    app_config: "AppConfig",
+) -> bool:
+    """Schedule a background review for ``thread_id``.
+
+    Returns True if a new task was started, False if queued or skipped.
+    """
+    evolution_cfg = app_config.skill_evolution
+    if not evolution_cfg.enabled or not evolution_cfg.review_worker.enabled:
+        return False
+
+    tool_count = extract_tool_call_count(messages)
+    if tool_count < evolution_cfg.review_worker.min_tool_calls:
+        logger.debug(
+            "skill_review_worker: skipping review for thread %s (%d tool calls < min %d)",
+            thread_id,
+            tool_count,
+            evolution_cfg.review_worker.min_tool_calls,
+        )
+        return False
+
+    lock = _get_lock(thread_id)
+    if lock.locked():
+        _pending[thread_id] = messages
+        logger.debug("skill_review_worker: queued review for thread %s (review in progress)", thread_id)
+        return False
+
+    asyncio.create_task(_run_review_with_pending(thread_id, messages, app_config))
+    logger.info("skill_review_worker: started review for thread %s (%d tool calls)", thread_id, tool_count)
+    return True

offline-backend-20260512/backend/tests/test_skill_review_worker.py

@@ -18,3 +18,85 @@ def test_review_worker_config_from_dict():
     assert cfg.review_worker.enabled is True
     assert cfg.review_worker.min_tool_calls == 5
     assert cfg.review_worker.model_name == "my-model"
+
+
+from unittest.mock import AsyncMock, MagicMock, patch
+import asyncio
+import pytest
+from langchain_core.messages import HumanMessage
+
+
+def _make_ai_message(tool_calls_data: list[dict]):
+    msg = MagicMock()
+    msg.__class__.__name__ = "AIMessage"
+    tcs = []
+    for tc in tool_calls_data:
+        m = MagicMock()
+        m.name = tc["name"]
+        m.args = tc.get("args", {})
+        tcs.append(m)
+    msg.tool_calls = tcs
+    return msg
+
+
+def test_extract_tool_call_count_counts_all():
+    from deerflow.skills.review_worker import extract_tool_call_count
+    msgs = [
+        HumanMessage(content="do something"),
+        _make_ai_message([{"name": "bash"}, {"name": "read_file"}]),
+        _make_ai_message([{"name": "write_file"}]),
+    ]
+    assert extract_tool_call_count(msgs) == 3
+
+
+def test_extract_tool_call_count_empty():
+    from deerflow.skills.review_worker import extract_tool_call_count
+    assert extract_tool_call_count([]) == 0
+
+
+def test_maybe_start_review_disabled_returns_false():
+    from deerflow.skills.review_worker import maybe_start_review
+    cfg = MagicMock()
+    cfg.skill_evolution.enabled = False
+    cfg.skill_evolution.review_worker.enabled = True
+    cfg.skill_evolution.review_worker.min_tool_calls = 2
+    msgs = [_make_ai_message([{"name": "bash"}, {"name": "read_file"}, {"name": "write_file"}])]
+    assert maybe_start_review("thread-1", msgs, cfg) is False
+
+
+def test_maybe_start_review_too_few_tool_calls_returns_false():
+    from deerflow.skills.review_worker import maybe_start_review
+    cfg = MagicMock()
+    cfg.skill_evolution.enabled = True
+    cfg.skill_evolution.review_worker.enabled = True
+    cfg.skill_evolution.review_worker.min_tool_calls = 5
+    msgs = [_make_ai_message([{"name": "bash"}])]
+    assert maybe_start_review("thread-4", msgs, cfg) is False
+
+
+def test_maybe_start_review_starts_task():
+    from deerflow.skills import review_worker as rw
+    # Reset module state so thread-3 has no lock
+    rw._locks.pop("thread-3", None)
+    rw._pending.pop("thread-3", None)
+
+    cfg = MagicMock()
+    cfg.skill_evolution.enabled = True
+    cfg.skill_evolution.review_worker.enabled = True
+    cfg.skill_evolution.review_worker.min_tool_calls = 2
+    cfg.skill_evolution.review_worker.timeout_seconds = 30
+    cfg.skill_evolution.review_worker.max_messages = 20
+    cfg.skill_evolution.review_worker.model_name = None
+
+    msgs = [_make_ai_message([{"name": "bash"}, {"name": "read_file"}, {"name": "write_file"}])]
+
+    with patch("deerflow.skills.review_worker.asyncio") as mock_asyncio:
+        fake_lock = MagicMock()
+        fake_lock.locked.return_value = False
+        mock_asyncio.Lock.return_value = fake_lock
+        mock_asyncio.create_task = MagicMock()
+
+        result = rw.maybe_start_review("thread-3", msgs, cfg)
+
+    assert result is True
+    mock_asyncio.create_task.assert_called_once()