orchestrata_ultimate.tools.conversation_tools

Reference › orchestrata_ultimate › tools › conversation_tools

These two tools give volks control over conversation capture (SPEC-021). Both wrap a ConversationQueueConsumer from the knowledge graph package, so they act on the Redis queue of conversation turns that is flushed into the graph. kernel.cli_main registers them at boot, and tests/tools/test_conversation_tools.py covers them.

ConvStatusTool reports the depth of the pending queue (REQ-021-216). ConvSyncTool flushes the queue into the graph through the consumer (REQ-021-217). Each is a BaseTool, takes its metadata and an optional tracer, and has a create classmethod that builds it over a shared consumer. ConvStatusTool.execute never raises; a Redis failure becomes a failed ToolResult. register_conversation_tools builds both over one consumer, registers them in a ToolRegistry and returns the pair.

At a glance

Class diagram

Dependency diagram

API

Classes

ConvStatusTool

Bases: BaseTool · Line 38

Reports the pending conversation-turn queue depth (REQ-021-216).

ConvSyncTool

Bases: BaseTool · Line 96

Flushes the conversation queue into the graph (REQ-021-217).

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/tools/conversation_tools.py

"""Conversation-capture volk tools — ConvStatusTool & ConvSyncTool (SPEC-021).

Two SPEC-062 ``BaseTool`` instances over the SAME injected
:class:`~orchestrata_ultimate.knowledge_graph.conversation_consumer.ConversationQueueConsumer`
(REQ-021-216/217): ``conv_status`` reports the pending queue depth,
``conv_sync`` drains it into the graph. Neither constructs its own
Redis/graph client — the boot seam (``kernel/cli_main.py``'s
``_boot_conversation_consumer``) builds the one shared consumer and passes
it to :func:`register_conversation_tools`, mirroring
``tools.task_tools.register_task_tools``'s shared-manager pattern.

Both tools never raise out of ``execute()``: a Redis/graph failure becomes a
failed :class:`ToolResult` (ERR-003 — non-blocking), matching every other
native tool in this package.
"""

from __future__ import annotations

from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.tracing.tracer import TraceConfig, Tracer
from orchestrata_ultimate.tools.base import (
    BaseTool,
    ToolCategory,
    ToolMetadata,
    ToolPriority,
    ToolResult,
)

if TYPE_CHECKING:
    from orchestrata_ultimate.knowledge_graph.conversation_consumer import (
        ConversationQueueConsumer,
    )
    from orchestrata_ultimate.tools.registry import ToolRegistry


class ConvStatusTool(BaseTool):
    """Reports the pending conversation-turn queue depth (REQ-021-216)."""

    def __init__(
        self,
        consumer: ConversationQueueConsumer,
        metadata: ToolMetadata,
        tracer: Tracer | None = None,
    ) -> None:
        self._consumer = consumer
        self._metadata = metadata
        self._tracer = tracer if tracer is not None else Tracer.default()

    @property
    def metadata(self) -> ToolMetadata:
        return self._metadata

    @traces(spec="021", story="US-021-007", reqs=["REQ-021-216"])
    async def execute(self, arguments: dict[str, Any]) -> ToolResult:
        """Return the queue depth. Never raises: a Redis failure becomes a
        failed :class:`ToolResult` (ERR-003), success at depth 0 included."""
        with self._tracer.span("ConvStatusTool.execute"):
            try:
                depth = self._consumer.depth()
            except Exception as exc:
                return ToolResult(
                    tool_name=self._metadata.name,
                    success=False,
                    output="",
                    error=str(exc),
                )
            return ToolResult(
                tool_name=self._metadata.name,
                success=True,
                output=f"{depth} conversation turn(s) pending",
            )

    @classmethod
    def create(
        cls,
        consumer: ConversationQueueConsumer,
        trace_config: TraceConfig | None = None,
    ) -> ConvStatusTool:
        """Build a :class:`ConvStatusTool` over the given shared *consumer*."""
        tracer = Tracer(trace_config) if trace_config is not None else Tracer.default()
        metadata = ToolMetadata(
            name="conv_status",
            description=(
                "Report the pending conversation-turn queue depth "
                "(SPEC-021 conversation capture)."
            ),
            category=ToolCategory.MEMORY,
            priority=ToolPriority.P2,
            tags=("conversation", "queue", "kg"),
        )
        return cls(consumer=consumer, metadata=metadata, tracer=tracer)


class ConvSyncTool(BaseTool):
    """Flushes the conversation queue into the graph (REQ-021-217)."""

    def __init__(
        self,
        consumer: ConversationQueueConsumer,
        metadata: ToolMetadata,
        tracer: Tracer | None = None,
    ) -> None:
        self._consumer = consumer
        self._metadata = metadata
        self._tracer = tracer if tracer is not None else Tracer.default()

    @property
    def metadata(self) -> ToolMetadata:
        return self._metadata

    @traces(spec="021", story="US-021-007", reqs=["REQ-021-217"])
    async def execute(self, arguments: dict[str, Any]) -> ToolResult:
        """Flush the queue via the consumer.

        ``success`` is ``False`` whenever any message failed to write
        (``report.failed > 0``) — the summary line always carries the
        synced/quarantined/failed counts either way, and this method itself
        never raises (the consumer's ``flush()`` already never does).
        """
        with self._tracer.span("ConvSyncTool.execute"):
            try:
                report = self._consumer.flush()
            except Exception as exc:
                return ToolResult(
                    tool_name=self._metadata.name,
                    success=False,
                    output="",
                    error=str(exc),
                )
            return ToolResult(
                tool_name=self._metadata.name,
                success=report.failed == 0,
                output=report.to_summary(),
            )

    @classmethod
    def create(
        cls,
        consumer: ConversationQueueConsumer,
        trace_config: TraceConfig | None = None,
    ) -> ConvSyncTool:
        """Build a :class:`ConvSyncTool` over the given shared *consumer*."""
        tracer = Tracer(trace_config) if trace_config is not None else Tracer.default()
        metadata = ToolMetadata(
            name="conv_sync",
            description=(
                "Flush pending conversation turns from the queue into the "
                "knowledge graph (SPEC-021 conversation capture)."
            ),
            category=ToolCategory.MEMORY,
            priority=ToolPriority.P2,
            tags=("conversation", "queue", "kg"),
        )
        return cls(consumer=consumer, metadata=metadata, tracer=tracer)


# ── wiring ────────────────────────────────────────────────────────────────


def register_conversation_tools(
    registry: ToolRegistry,
    consumer: ConversationQueueConsumer,
    trace_config: TraceConfig | None = None,
) -> tuple[ConvStatusTool, ConvSyncTool]:
    """Build :class:`ConvStatusTool` + :class:`ConvSyncTool` sharing one
    :class:`ConversationQueueConsumer`, and register both into ``registry``
    (mirrors ``tools.task_tools.register_task_tools``).
    """
    status_tool = ConvStatusTool.create(consumer, trace_config)
    sync_tool = ConvSyncTool.create(consumer, trace_config)
    registry.register(status_tool, replace=True)
    registry.register(sync_tool, replace=True)
    return status_tool, sync_tool