Merge pull request #215 from zts212653/feat/issue-201-guidance-injection

feat(guidance): in-flight user guidance injection

Author: bouillipx

client/main.py

@@ -295,6 +295,9 @@ async def _read_control_stdin_line() -> str | None:
     ResourceAction.MCP_RELOAD,
     ResourceAction.MCP_SHOW_TOOL,
     ResourceAction.SKILLS_LIST,
+    ResourceAction.GUIDE_INJECT,
+    ResourceAction.GUIDE_LIST,
+    ResourceAction.GUIDE_CLEAR,
 )
 _SESSION_RESUME_ACTION = "session:resume"
 

dare_framework/agent/builder.py

@@ -83,6 +83,8 @@
 from dare_framework.transport.interaction.dispatcher import ActionHandlerDispatcher
 from dare_framework.transport.kernel import AgentChannel
 from dare_framework.context.manage_context import ManageContextTool, MANAGE_CONTEXT_TOOL_NAME
+from dare_framework.guidance import GuidanceQueue
+from dare_framework.guidance.action_handler import GuidanceActionHandler
 
 logger = logging.getLogger(__name__)
 
@@ -359,11 +361,13 @@ async def build(self) -> TAgent:
             tools,
             self._tool_providers,
         )
+        guidance_queue = GuidanceQueue()
         context = self._build_context(
             config=config,
             knowledge=knowledge,
             sys_prompt=sys_prompt,
             tool_gateway=tool_gateway,
+            guidance_queue=guidance_queue,
         )
         self._context = context
         agent = self._build_impl(
@@ -385,6 +389,7 @@ async def build(self) -> TAgent:
             )
             action_dispatcher.register_action_handler(ToolsActionHandler(tool_manager))
             action_dispatcher.register_action_handler(ApprovalsActionHandler(approval_manager))
+            action_dispatcher.register_action_handler(GuidanceActionHandler(guidance_queue))
             if skill_store is not None:
                 action_dispatcher.register_action_handler(SkillsActionHandler(skill_store))
             action_dispatcher.register_action_handler(
@@ -567,6 +572,7 @@ def _build_context(
             knowledge: IKnowledge | None,
             sys_prompt: Prompt | None,
             tool_gateway: IToolGateway | None,
+            guidance_queue: GuidanceQueue | None = None,
     ) -> Context:
         """Build context with shared defaults for all builder variants.
 
@@ -586,6 +592,7 @@ def _build_context(
             tool_gateway=tool_gateway,
             assemble_context=self._assemble_context,
             context_window_tokens=self._context_window_tokens,
+            guidance_queue=guidance_queue,
         )
         # 若 builder 已解析到 model：
         # - 默认（compression 为空或其他值）视为启用 "moving" 压缩；

dare_framework/context/context.py

@@ -12,11 +12,12 @@
 
 if TYPE_CHECKING:
     from dare_framework.compression.moving_compression import MovingCompressor
+    from dare_framework.guidance.guidance_queue import GuidanceQueue
     from dare_framework.model.types import Prompt
     from dare_framework.skill.types import Skill
 
 from dare_framework.context.kernel import IContext, IRetrievalContext, IAssembleContext
-from dare_framework.context.types import AssembledContext, Budget, Message, MessageRole
+from dare_framework.context.types import AssembledContext, Budget, Message, MessageKind, MessageRole
 
 
 # ============================================================
@@ -45,6 +46,7 @@ def __init__(
             assemble_context: IAssembleContext | None = None,
             context_window_tokens: int | None = None,
             moving_compressor: MovingCompressor | None = None,
+            guidance_queue: GuidanceQueue | None = None,
     ) -> None:
         if config is None:
             raise ValueError("Context requires a non-null Config")
@@ -69,6 +71,9 @@ def __init__(
         # 可选：移动窗口压缩器，通过 moving_compressor.prune(self) 使用
         self._moving_compressor = moving_compressor
 
+        # 可选：用户引导队列，在 assemble() 时自动排空注入到 STM
+        self._guidance_queue: GuidanceQueue | None = guidance_queue
+
         # Initialize default short-term memory if not provided
         if self._short_term_memory is None:
             from dare_framework.memory.in_memory_stm import InMemorySTM
@@ -198,7 +203,26 @@ def list_tools(self) -> list[CapabilityDescriptor]:
 
     # ========== Assembly Methods (Core) ==========
 
+    def _drain_guidance(self) -> None:
+        """Drain pending guidance items into STM.
+
+        Called before compression and assembly so that guidance messages
+        participate in token-budget pruning and appear in the assembled
+        context exactly once.
+        """
+        if self._guidance_queue is None:
+            return
+        items = self._guidance_queue.drain_all_sync()
+        for item in items:
+            self.stm_add(Message(
+                role=MessageRole.USER,
+                kind=MessageKind.CHAT,
+                text=item.content,
+                metadata={"guidance_id": item.id, "source": "user_guidance"},
+            ))
+
     def assemble(self) -> AssembledContext:
+        self._drain_guidance()
         return self._assemble_context.assemble(self)
 
     async def compress(self, **options: Any) -> None:
@@ -214,7 +238,13 @@ async def compress(self, **options: Any) -> None:
         await self._moving_compressor.prune(self, **prune_opts)
 
     async def assemble_for_model(self, **options: Any) -> AssembledContext:
-        """供模型调用的装配入口：在内部静默触发压缩，然后返回 AssembledContext。"""
+        """供模型调用的装配入口：先排空引导队列，再压缩，最后装配。
+
+        Guidance is drained **before** compression so that guidance messages
+        participate in token-budget pruning and do not silently push the
+        final prompt over the configured context window.
+        """
+        self._drain_guidance()
         await self.compress(**options)
         return self.assemble()
 

dare_framework/context/smartcontext.py

@@ -10,6 +10,7 @@
 
 if TYPE_CHECKING:
     from dare_framework.config.types import Config
+    from dare_framework.guidance.guidance_queue import GuidanceQueue
     from dare_framework.model.types import Prompt
     from dare_framework.skill.types import Skill
     from dare_framework.tool.kernel import IToolGateway
@@ -38,6 +39,7 @@ def __init__(
         skill: Skill | None = None,
         assemble_context: IAssembleContext | None = None,
         context_window_tokens: int | None = None,
+        guidance_queue: GuidanceQueue | None = None,
     ) -> None:
         # SmartContext 默认使用带 id/mark 能力的 InMemorySmartSTM，
         # 仅当外部显式传入 short_term_memory 时才使用外部实现。
@@ -56,6 +58,7 @@ def __init__(
             skill=skill,
             assemble_context=assemble_context,
             context_window_tokens=context_window_tokens,
+            guidance_queue=guidance_queue,
         )
 
     # ========== Smart Extensions: Core / TaskComplete / STM Remove ==========

dare_framework/guidance/__init__.py

@@ -0,0 +1,5 @@
+"""Guidance injection — in-flight user guidance queue and action handler."""
+
+from dare_framework.guidance.guidance_queue import GuidanceItem, GuidanceQueue
+
+__all__ = ["GuidanceItem", "GuidanceQueue"]

dare_framework/guidance/action_handler.py

@@ -0,0 +1,63 @@
+"""Guidance-domain deterministic action handler."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from dare_framework.guidance.guidance_queue import GuidanceItem, GuidanceQueue
+from dare_framework.transport.interaction.handlers import IActionHandler
+from dare_framework.transport.interaction.resource_action import ResourceAction
+
+
+class GuidanceActionHandler(IActionHandler):
+    """Handle deterministic guidance-domain actions (inject / list / clear)."""
+
+    def __init__(self, guidance_queue: GuidanceQueue) -> None:
+        self._queue = guidance_queue
+
+    def supports(self) -> set[ResourceAction]:
+        return {
+            ResourceAction.GUIDE_INJECT,
+            ResourceAction.GUIDE_LIST,
+            ResourceAction.GUIDE_CLEAR,
+        }
+
+    # noinspection PyMethodOverriding
+    async def invoke(
+        self,
+        action: ResourceAction,
+        **params: Any,
+    ) -> Any:
+        if action == ResourceAction.GUIDE_INJECT:
+            content = params.get("content")
+            if not content or not str(content).strip():
+                raise ValueError("guide:inject requires a non-empty 'content' parameter")
+            item = self._queue.enqueue(str(content))
+            return {
+                "id": item.id,
+                "pending_count": self._queue.pending_count,
+            }
+
+        if action == ResourceAction.GUIDE_LIST:
+            items = self._queue.peek_all()
+            return {
+                "items": [_item_to_dict(i) for i in items],
+                "count": len(items),
+            }
+
+        if action == ResourceAction.GUIDE_CLEAR:
+            removed = self._queue.clear()
+            return {"removed": removed}
+
+        raise ValueError(f"unsupported guidance action: {action.value}")
+
+
+def _item_to_dict(item: GuidanceItem) -> dict[str, Any]:
+    return {
+        "id": item.id,
+        "content": item.content,
+        "created_at": item.created_at,
+    }
+
+
+__all__ = ["GuidanceActionHandler"]

dare_framework/guidance/guidance_queue.py

@@ -0,0 +1,94 @@
+"""Thread-safe guidance queue with consume-on-drain semantics."""
+
+from __future__ import annotations
+
+import threading
+import time
+import uuid
+from collections import deque
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class GuidanceItem:
+    """Single guidance entry enqueued by the user."""
+
+    id: str
+    content: str
+    created_at: float
+
+
+class GuidanceQueue:
+    """FIFO queue for in-flight user guidance messages.
+
+    The queue is written to by ``GuidanceActionHandler`` (via ``enqueue``) and
+    drained by ``Context.assemble()`` (via ``drain_all_sync``).  A
+    ``threading.Lock`` is used so that ``drain_all_sync`` can be called from
+    the synchronous ``assemble()`` method.
+    """
+
+    def __init__(self, *, max_size: int = 100) -> None:
+        self._items: deque[GuidanceItem] = deque()
+        self._lock = threading.Lock()
+        self._max_size = max_size
+
+    # ------------------------------------------------------------------
+    # Write side (called from async action handler → GIL-safe)
+    # ------------------------------------------------------------------
+
+    def enqueue(self, content: str) -> GuidanceItem:
+        """Append a guidance message.  Returns the created item for auditing.
+
+        Raises ``ValueError`` if *content* is empty or whitespace-only, or if
+        the queue has reached *max_size*.
+        """
+        if not content or not content.strip():
+            raise ValueError("guidance content must be a non-empty string")
+        item = GuidanceItem(
+            id=str(uuid.uuid4()),
+            content=content,
+            created_at=time.time(),
+        )
+        with self._lock:
+            if len(self._items) >= self._max_size:
+                raise RuntimeError(
+                    f"guidance queue full ({self._max_size} items); "
+                    "drain or clear before enqueuing more"
+                )
+            self._items.append(item)
+        return item
+
+    # ------------------------------------------------------------------
+    # Read / consume side
+    # ------------------------------------------------------------------
+
+    def drain_all_sync(self) -> list[GuidanceItem]:
+        """Pop **all** pending items atomically.
+
+        This is the only consumption path — called from the synchronous
+        ``Context.assemble()`` method.  After this call the queue is empty.
+        """
+        with self._lock:
+            items = list(self._items)
+            self._items.clear()
+        return items
+
+    def peek_all(self) -> list[GuidanceItem]:
+        """Return a snapshot of pending items without consuming them."""
+        with self._lock:
+            return list(self._items)
+
+    def clear(self) -> int:
+        """Remove all pending items.  Returns the number removed."""
+        with self._lock:
+            count = len(self._items)
+            self._items.clear()
+        return count
+
+    @property
+    def pending_count(self) -> int:
+        """Number of pending guidance items (lock-free; approximate)."""
+        return len(self._items)
+
+
+__all__ = ["GuidanceItem", "GuidanceQueue"]

dare_framework/transport/interaction/resource_action.py

@@ -24,6 +24,9 @@ class ResourceAction(StrEnum):
     MCP_SHOW_TOOL = "mcp:show-tool"
     SKILLS_LIST = "skills:list"
     MODEL_GET = "model:get"
+    GUIDE_INJECT = "guide:inject"
+    GUIDE_LIST = "guide:list"
+    GUIDE_CLEAR = "guide:clear"
 
     @classmethod
     def value_of(cls, raw: str) -> ResourceAction | None: