orchestrata_ultimate.agents.volks.concierge.volk

Reference › orchestrata_ultimate › agents › volks › concierge › volk

ConciergeVolk (alias U:Cx, role Concierge) is the front door of the delivery pipeline (specification 032). It receives the raw human intent, runs a bounded clarification dialogue, and produces the requirements bundle that the Analyst (agent) receives. It extends BaseAgent (base); the only importer in the reference is the kernel pipeline. ConciergeAgent is an alias subclass. The curated role description is in volks.

What it receives: an intent packet whose payload text is the human's request, and later response packets carrying the human's answers. What it produces: clarification packets addressed to HUMAN, and finally an analyze_requirements packet addressed to U:BA.

_handle_intent resets the conversation state, records the original intent and asks the LLM, with CONCIERGE_SYSTEM_PROMPT, for clarifying questions. It never finalises on the first call: if the model answers with a completion marker, the reply is replaced by a fixed set of three questions. _handle_response appends the human's answers, increments a round counter and asks the model whether to continue. Requirements are finalised when the model signals completion, when the human's text already contains the completion marker, or unconditionally once the round counter reaches MAX_QA_ROUNDS, which is 4. _finalize parses the labelled fields from the compiled text: PROLOG_FACTS lines, ESCALATE_TO, TASK_SUMMARY, PROJECT_NAME and WORK_TYPE (checked against a fixed set of work types, default feature). Missing fields do not raise. detect_ui_requirement scans the texts for UI keywords such as "web app", "dashboard" and "frontend", and the flag travels in the handoff payload, which also carries the conversation history, the original intent and the task summary. The destination is fixed as U:BA; the parsed ESCALATE_TO value is recorded but does not change it.

At a glance

Class diagram

Dependency diagram

API

Classes

ConciergeVolk (pydantic)

Bases: BaseAgent · Decorators: traces(spec='032', story='US-032-001', reqs=['REQ-032-001', 'REQ-032-002', 'REQ-032-003', 'REQ-032-005', 'REQ-032-048',… · Line 119

ConciergeAgent (pydantic)

Bases: ConciergeVolk · Line 460

Agent alias for ConciergeVolk.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/agents/volks/concierge/volk.py

# mypy: ignore-errors
"""U:Cx — Concierge volk with real LLM brain."""

import logging
from typing import ClassVar

from pydantic import Field

from orchestrata_ultimate.agents.volks.base import BaseAgent
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph_directives import (
    extract_graph_context,
)
from orchestrata_ultimate.core.packet import CXPPacket, WorkStatus

logger = logging.getLogger(__name__)

# REQ-032-048 through REQ-032-052: CONCIERGE_SYSTEM_PROMPT is defined as a module-level string constant
# that instructs the LLM to elicit requirements across seven dimensions (D1-D7), signal REQUIREMENTS_COMPLETE
# when dimensions are sufficiently covered, emit PROLOG_FACTS in valid Prolog syntax, and emit
# TASK_SUMMARY, PROJECT_NAME, and ESCALATE_TO on finalization
CONCIERGE_SYSTEM_PROMPT = """You are U:Cx, the Concierge — a senior business analyst \
and requirements engineer for the Orchestrata multi-agent system.

YOUR MISSION: Transform a vague human intent into a precise, testable, \
implementation-ready requirements specification that a software architect \
can act on without ambiguity.

PROCESS (follow in order):
1. Receive the human's intent (what they want to build).
2. Perform a structured gap analysis using the REQUIREMENTS DIMENSIONS below.
3. ALWAYS ask 3-6 clarifying questions on the first response — never skip \
   to REQUIREMENTS_COMPLETE on the first call, even if the intent seems complete.
4. After receiving answers, either ask follow-up questions (if critical gaps \
   remain) or compile the final requirements specification.

REQUIREMENTS DIMENSIONS — check each one for completeness:
  D1. FUNCTIONAL SCOPE: What features must the system provide? What are the \
      inputs, outputs, and core workflows? What is explicitly OUT of scope?
  D2. TECHNOLOGY STACK: Language, framework, database, message broker. If the \
      human specifies these, use them exactly. If unspecified, recommend \
      Python 3.14+ with stdlib/minimal dependencies.
  D3. DATA MODEL: What entities exist? What are their relationships? What \
      persistence is needed (ephemeral, file, database)?
  D4. INTERFACES: CLI, REST API, WebSocket, library? What are the external \
      integration points?
  D5. QUALITY ATTRIBUTES: Performance targets, error handling strategy, \
      testing requirements, logging/observability needs.
  D6. DEPLOYMENT: Default is Incus containers (Linux). If the human says \
      Docker, use Docker. If unspecified, use Incus. Never assume Docker.
  D7. CONSTRAINTS: Budget, timeline, team size, existing codebase to \
      integrate with, regulatory requirements.

QUESTION STRATEGY (based on requirements elicitation research):
- Ask questions that disambiguate, not questions that educate.
- Each question must target ONE specific gap in ONE dimension above.
- Phrase questions as forced choices when possible ("Do you need X or Y?" \
  not "What do you need?") — constrained options reduce ambiguity.
- For scope: ask about boundaries ("Should the system handle X, or is that \
  out of scope?").
- For data: ask about cardinality and lifecycle ("Is there one X per Y, or \
  many? Are they persistent or ephemeral?").
- Never ask generic questions like "Any other requirements?" — be specific.

WHEN ASKING QUESTIONS, output EXACTLY in this format:
QUESTIONS:
1. [D<n>] <specific question targeting dimension n>
2. [D<n>] <specific question targeting dimension n>
...

WHEN REQUIREMENTS ARE COMPLETE, output EXACTLY in this format:
REQUIREMENTS_COMPLETE:
- requirement_key: requirement_value
- requirement_key: requirement_value
...

PROLOG_FACTS:
fact(requirement, key, value).
fact(requirement, key, value).
...

ESCALATE_TO: U:BA
TASK_SUMMARY: <one-line summary of what to build>
PROJECT_NAME: <short kebab-case name for the project directory, 2-4 words \
max, e.g. "redis-task-queue" or "async-event-bus">
WORK_TYPE: <one of: feature | refactor | test | documentation | analysis | \
fix | review | spec | dependency | performance>

HANDOFF: On finalization your REQUIREMENTS_COMPLETE specification is handed to \
U:BA as an "analyze_requirements" message, so U:BA can classify the work and \
write user stories directly from it without re-asking the human.

REQUIREMENTS_COMPLETE OUTPUT RULES:
- Every requirement must be testable — it should be possible to write a \
  pass/fail test for it.
- Group requirements by dimension (FUNCTIONAL, DATA, INTERFACE, QUALITY, \
  DEPLOYMENT).
- Include both functional requirements AND non-functional requirements.
- Include explicit OUT OF SCOPE items to prevent scope creep downstream.
- Include error handling requirements ("When X fails, the system must Y").
"""


@traces(
    spec="032",
    story="US-032-001",
    reqs=[
        "REQ-032-001",  # ConciergeVolk is defined as a class extending BaseAgent
        "REQ-032-002",  # alias is set to the string constant "U:Cx"
        "REQ-032-003",  # role is set to the string constant "Concierge (Intake)"
        "REQ-032-005",  # ConciergeVolk exposes MAX_QA_ROUNDS as a class-level integer constant equal to 4
        "REQ-032-048",  # CONCIERGE_SYSTEM_PROMPT is defined as a module-level string constant
        "REQ-032-049",  # Prompt instructs LLM to elicit requirements across D1-D7 dimensions
        "REQ-032-050",  # Prompt instructs LLM to signal REQUIREMENTS_COMPLETE
        "REQ-032-051",  # Prompt instructs LLM to emit PROLOG_FACTS in valid Prolog syntax
        "REQ-032-052",  # Prompt instructs LLM to emit TASK_SUMMARY, PROJECT_NAME, and ESCALATE_TO
    ],
)
class ConciergeVolk(BaseAgent):
    alias: str = Field(default="U:Cx")
    role: str = "Concierge (Intake)"

    # G213: Max Q&A rounds before forced finalization
    MAX_QA_ROUNDS: ClassVar[int] = 4

    @traces(
        spec="032",
        story="US-032-001",
        reqs=[
            "REQ-032-006",  # conversation_history is initialised as an empty list on instantiation
            "REQ-032-007",  # original_intent is initialised as None or empty string on instantiation
            "REQ-032-008",  # prolog_facts is initialised as an empty string on instantiation
            "REQ-032-009",  # round_count is initialised to 0 on instantiation
            "REQ-032-010",  # All instance state attributes persist for the lifetime of the handling Volk instance
        ],
    )
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.conversation_history = []
        self.requirements = {}
        self.prolog_facts = []
        self.escalation_target = None
        self.task_summary = None
        self.project_name = None
        self.original_intent = ""
        self._qa_round = 0
        self.work_type = "feature"  # Default, overridden during finalization
        self._seen_answers: set = set()
        # Propagates the established "project_dir" contract key (the
        # pipeline_runner.py first-stage seam; QA/compliance volks honor it
        # on chained stages) — without it U:Cp falls back to CWD and scans
        # the wrong tree.
        self.project_dir = ""

    #: Keywords that mark an intent as needing a user interface (drives the
    #: ``has_ui`` flag → whether the U:Ux stage runs).
    UI_KEYWORDS: ClassVar[tuple[str, ...]] = (
        "web app",
        "dashboard",
        "frontend",
        "ui",
        "user interface",
        "react",
        "vue",
        "htmx",
        "html",
        "css",
        "browser",
        "responsive",
    )

    @staticmethod
    @traces(spec="032", story="US-004", reqs=["REQ-032-041"])
    def detect_ui_requirement(texts: list[str]) -> bool:
        """Return True if any UI keyword appears across ``texts`` (FR-009).

        Pure over its inputs so the caller can pass the full conversation
        (``[llm_response, original_intent, *conversation_history]``) — a UI need
        stated only in a mid-conversation answer is caught, not just one in the
        original intent or the final compiled response.
        """
        combined = " ".join(t for t in texts if t).lower()
        return any(keyword in combined for keyword in ConciergeVolk.UI_KEYWORDS)

    @traces(
        spec="032",
        story="US-032-001",
        reqs=[
            "REQ-032-011",  # process(packet) inspects the CXP packet subject field for routing
            "REQ-032-012",  # Packets with subject "intent" are routed to _handle_intent()
            "REQ-032-013",  # Packets with subject "response" are routed to _handle_response()
            "REQ-032-014",  # Packets with unrecognised subjects are rejected with a logged error
        ],
    )
    async def process(self, packet: CXPPacket) -> CXPPacket | None:
        self._graph_context = extract_graph_context(packet.payload)
        self.project_dir = str(packet.payload.get("project_dir", "") or "")
        if packet.subject == "intent":
            return await self._handle_intent(packet)
        elif packet.subject == "response":
            return await self._handle_response(packet)
        # REQ-032-014 / ERR-003: an unrecognised subject is dropped — but not
        # silently. Log it so an unwired/misrouted packet is diagnosable instead
        # of vanishing (the module had no logger before).
        logger.warning(
            "U:Cx dropping packet with unrecognised subject %r (thread=%s, from=%s)",
            packet.subject,
            packet.thread_id,
            packet.from_agent,
        )
        return None

    @traces(
        spec="032",
        story="US-032-001",
        reqs=[
            "REQ-032-016",  # _handle_intent() stores the packet body as original_intent on the first call
            "REQ-032-017",  # _handle_intent() appends the human intent message to conversation_history
            "REQ-032-018",  # _handle_intent() calls the LLM with CONCIERGE_SYSTEM_PROMPT and current conversation_history
            "REQ-032-019",  # _handle_intent() routes the LLM response to HUMAN as clarifying questions
            "REQ-032-020",  # _handle_intent() never compiles or finalises requirements on the first call (G162-1)
        ],
    )
    async def _handle_intent(self, packet: CXPPacket) -> CXPPacket:
        self.status = WorkStatus.IN_PROGRESS
        self.current_thread = packet.thread_id
        self.summary = "Gathering requirements"
        self._qa_round = 0  # Reset counter for new intent
        self.conversation_history = []  # Reset history

        intent_text = packet.payload.get("text", "")
        self.original_intent = intent_text
        self.conversation_history.append(f"Human intent: {intent_text}")

        prompt = f"The human wants to: {intent_text}\n\nAnalyze this intent and ask clarifying questions."
        response_text = await self._call_llm(
            CONCIERGE_SYSTEM_PROMPT,
            prompt,
            temperature=1.3,
            graph_context=self._graph_context,
        )
        self.conversation_history.append(f"U:Cx response: {response_text}")

        # G162-1: Never fast-path on first call — runner needs the
        # HUMAN packet to extract thread_id for approve flow
        if "REQUIREMENTS_COMPLETE:" in response_text:
            response_text = (
                "QUESTIONS:\n"
                "1. What deployment platform? (Incus, Docker, bare metal)\n"
                "2. Any specific testing requirements?\n"
                "3. Any performance or scale constraints?\n"
            )

        # Extract questions and send to human
        return CXPPacket(
            from_agent="U:Cx",
            to_agent="HUMAN",
            thread_id=packet.thread_id,
            subject="clarification",
            payload={
                "text": response_text,
                "raw": response_text,
                "project_dir": self.project_dir,
            },
            work_status=WorkStatus.IN_PROGRESS,
        )

    @traces(
        spec="032",
        story="US-032-001",
        reqs=[
            "REQ-032-022",  # _handle_response() appends the human answer to conversation_history before the LLM call
            "REQ-032-023",  # _handle_response() increments round_count before LLM invocation
            "REQ-032-024",  # _handle_response() calls the LLM with the full conversation_history each time
            "REQ-032-025",  # _handle_response() checks for REQUIREMENTS_COMPLETE signal in LLM response
            "REQ-032-026",  # _handle_response() checks whether round_count >= MAX_QA_ROUNDS (G213)
            "REQ-032-027",  # If REQUIREMENTS_COMPLETE is signalled or MAX_QA_ROUNDS is reached, _finalize() is called
            "REQ-032-028",  # If neither condition is met, LLM response is sent to HUMAN as follow-up questions
            "REQ-032-030",  # When round_count >= MAX_QA_ROUNDS, _finalize() is called unconditionally regardless of LLM signal
            "REQ-032-031",  # No additional questions are sent to HUMAN after forced finalization is triggered
            "REQ-032-032",  # Forced finalization compiles requirements from all conversation_history accumulated to that point
        ],
    )
    async def _handle_response(self, packet: CXPPacket) -> CXPPacket:
        # Replay prior human answers into history once. The kernel resends the
        # full prior_answers list each round; dedup against _seen_answers so an
        # answer already recorded is not appended (and re-analysed) again.
        for answer in packet.payload.get("prior_answers", []):
            if answer and answer not in self._seen_answers:
                self.conversation_history.append(f"Human response: {answer}")
                self._seen_answers.add(answer)

        response_text = packet.payload.get("text", "")

        # If human explicitly signals completion, force-finalize immediately
        # Do NOT call LLM again — it may return QUESTIONS causing a loop
        if "REQUIREMENTS_COMPLETE" in response_text:
            self.conversation_history.append(f"Human response: {response_text}")
            history = "\n".join(self.conversation_history)
            llm_response = (
                f"REQUIREMENTS_COMPLETE:\n{history}\n\n"
                f"TASK_SUMMARY: {self.original_intent}\n"
                f"ESCALATE_TO: U:BA"
            )
            return self._finalize(packet, llm_response)

        self._qa_round += 1
        self.conversation_history.append(f"Human response: {response_text}")

        # G213: Force finalization after MAX_QA_ROUNDS
        # Do NOT call LLM again — finalize with conversation history directly
        if self._qa_round >= self.MAX_QA_ROUNDS:
            history = "\n".join(self.conversation_history)
            llm_response = (
                f"REQUIREMENTS_COMPLETE:\n{history}\n\n"
                f"TASK_SUMMARY: {self.original_intent}\n"
                f"ESCALATE_TO: U:BA"
            )
            return self._finalize(packet, llm_response)

        history = "\n".join(self.conversation_history)

        prompt = (
            f"Conversation so far:\n{history}\n\n"
            f"Human's response: {response_text}\n\n"
            f"This is round {self._qa_round} of {self.MAX_QA_ROUNDS}.\n"
            "Based on this response, either:\n"
            "1. Ask MORE questions ONLY for dimensions not yet covered "
            "(use QUESTIONS: format). Do NOT repeat earlier questions.\n"
            "2. If all 7 dimensions are sufficiently covered, compile "
            "requirements NOW (use REQUIREMENTS_COMPLETE: format).\n\n"
            "If this is round 3+, strongly prefer REQUIREMENTS_COMPLETE."
        )
        llm_response = await self._call_llm(
            CONCIERGE_SYSTEM_PROMPT,
            prompt,
            temperature=0.7,
            graph_context=self._graph_context,
        )
        self.conversation_history.append(f"U:Cx response: {llm_response}")

        if "REQUIREMENTS_COMPLETE:" in llm_response:
            return self._finalize(packet, llm_response)

        return CXPPacket(
            from_agent="U:Cx",
            to_agent="HUMAN",
            thread_id=packet.thread_id,
            subject="clarification",
            payload={
                "text": llm_response,
                "raw": llm_response,
                "project_dir": self.project_dir,
            },
            work_status=WorkStatus.IN_PROGRESS,
        )

    @traces(
        spec="032",
        story="US-032-001",
        reqs=[
            "REQ-032-035",  # _finalize() parses PROLOG_FACTS block from LLM response and stores it in prolog_facts
            "REQ-032-036",  # _finalize() parses ESCALATE_TO field from LLM response
            "REQ-032-037",  # _finalize() parses TASK_SUMMARY paragraph from LLM response
            "REQ-032-038",  # _finalize() parses PROJECT_NAME field from LLM response
            "REQ-032-039",  # Missing or malformed structured fields are handled without raising an exception
            "REQ-032-040",  # All parsed fields are included in the outbound CXP packet to U:BA
            "REQ-032-042",  # Detected UI requirement flag is included in the handoff packet to U:BA
            "REQ-032-043",  # UI detection is applied at finalization time, not during elicitation rounds
            "REQ-032-044",  # After finalization, a CXP packet is always sent to U:BA
            "REQ-032-045",  # The outbound packet to U:BA includes TASK_SUMMARY, PROJECT_NAME, PROLOG_FACTS, conversation_history, and UI flag
            "REQ-032-046",  # ESCALATE_TO is parsed and logged but U:BA is the fixed pipeline destination
            "REQ-032-047",  # No other Volk receives the requirements packet directly from U:Cx
        ],
    )
    def _finalize(self, packet: CXPPacket, llm_response: str) -> CXPPacket:
        """Parse requirements, emit Prolog facts, escalate."""
        self.output_artifacts.append(llm_response)

        # Parse PROLOG_FACTS if present
        if "PROLOG_FACTS:" in llm_response:
            facts_section = llm_response.split("PROLOG_FACTS:")[1]
            if "ESCALATE_TO:" in facts_section:
                facts_section = facts_section.split("ESCALATE_TO:")[0]
            for line in facts_section.strip().split("\n"):
                line = line.strip()
                if line and line.startswith("fact("):
                    self.prolog_facts.append(line.rstrip("."))

        # Parse escalation target
        if "ESCALATE_TO:" in llm_response:
            esc_line = llm_response.split("ESCALATE_TO:")[1].split("\n")[0].strip()
            self.escalation_target = esc_line

        # Parse task summary
        if "TASK_SUMMARY:" in llm_response:
            summary_line = llm_response.split("TASK_SUMMARY:")[1].split("\n")[0].strip()
            self.task_summary = summary_line

        # Parse project name
        if "PROJECT_NAME:" in llm_response:
            name_line = llm_response.split("PROJECT_NAME:")[1].split("\n")[0].strip()
            self.project_name = name_line

        # Parse work type for pipeline routing
        self.work_type = "feature"  # Default
        if "WORK_TYPE:" in llm_response:
            wt = llm_response.split("WORK_TYPE:")[1].split("\n")[0].strip().lower()
            valid_types = {
                "feature",
                "refactor",
                "test",
                "documentation",
                "analysis",
                "fix",
                "review",
                "spec",
                "dependency",
                "performance",
            }
            if wt in valid_types:
                self.work_type = wt

        self.status = WorkStatus.RESOLVED
        self.summary = "Requirements gathered — escalating"

        # Detect UI requirement across the WHOLE conversation, not just the
        # final compiled response + original intent (FR-009 / REQ-032-041): a UI
        # keyword the human gives only in a later clarification answer lives in
        # conversation_history, so a mid-conversation "make it a dashboard" is no
        # longer missed (which would wrongly skip the U:Ux stage).
        has_ui = self.detect_ui_requirement(
            [llm_response, self.original_intent, *self.conversation_history]
        )

        # Always route to BA first (BA decomposes into stories, then PM scopes)
        # Ignore LLM's ESCALATE_TO — pipeline order is fixed: BA → PM → Ar → Cd
        target = "U:BA"
        return CXPPacket(
            from_agent="U:Cx",
            to_agent=target,
            thread_id=packet.thread_id,
            subject="analyze_requirements",
            payload={
                "requirements": llm_response,
                "conversation_history": list(self.conversation_history),
                "original_intent": self.original_intent,
                "task_summary": self.task_summary or "Implement as specified",
                "project_name": self.project_name,
                "prolog_facts": self.prolog_facts,
                "has_ui": has_ui,
                "work_type": self.work_type,
                "project_dir": self.project_dir,
            },
            work_status=WorkStatus.PENDING,
        )


# Class alias for orchestrata-ultimate naming
class ConciergeAgent(ConciergeVolk):
    """Agent alias for ConciergeVolk."""

    pass