request_plan.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/request_plan.py
 
"""
Provider-neutral request IR — core builds it, the provider renders it (plan 135 §3.4).
 
Purpose
-------
Decouple *what* to send (a provider-neutral :class:`ProviderRequestPlan`) from
*how* a given provider shapes the wire request (:class:`RenderedRequest`,
produced by ``ProviderPlugin.render_request``). For chat-completions providers
(DeepSeek/ollama/example) ``session.messages[0]`` stays the single source of
truth: the default renderer reproduces it **byte-for-byte**. The provider seam
exists so the future Anthropic bundle (plan 137) can rebuild native ``system``
blocks from the same plan without touching the inner loop.
 
Responsibilities
----------------
- :class:`ProviderRequestPlan` — the frozen, provider-neutral request IR.
- :class:`RenderedRequest` — the rendered shape ``render_request`` returns.
- :func:`build_request_plan` — decompose a chat ``messages`` list (by sentinel,
  not index) into the IR at the request-assembly point.
- :func:`compose_chat_messages` — the inverse used by the default OpenAI-Chat
  renderer; ``policy_prompt + tool_reference`` reproduces ``messages[0]`` exactly.
 
Diagnostics
-----------
Domain: OC_CHAT.CLIENT
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- :func:`build_request_plan` followed by :func:`compose_chat_messages` is a
  byte-identical round-trip of the chat ``messages`` list (``messages[0]`` is
  split by pure slicing at the ``\\n\\n## Tools`` boundary, so concatenation
  always reconstructs the original — even for ``/system``-customised prompts).
- The dataclasses are frozen; building a plan never mutates the input messages.
- This module imports no other ``oc_chat`` runtime module at load time (the
  variable-context sentinel is imported lazily) to stay cycle-free.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.session (lazy; VARIABLE_CONTEXT_SENTINEL to find the variable-context message)
"""
from __future__ import annotations
 
from dataclasses import dataclass, field
 
from oc_diagnostics import _dbg
 
_dbg("OC_CHAT.CLIENT", "request_plan", "module", 4)
 
#: messages[0] boundary between the policy/persona and the tool-use section.
#: Both the tools-present and no-tools sections open with ``## Tools`` (policy.py),
#: and ``build_system_prompt`` joins them with a blank line, so this marker
#: locates the split. Pure slicing, so concatenation always rebuilds the original.
_TOOL_SECTION_MARKER = "\n\n## Tools"
 
 
@dataclass(frozen=True)
class ProviderRequestPlan:
    """Provider-neutral request IR (plan 135 §3.4)."""
 
    model: str
    policy_prompt: str                       # Option C policy/persona (provider-neutral)
    tool_reference: str                      # messages[0] tool-use section (preamble+map, or no-tools policy; "" if absent)
    variable_context: str                    # messages[1] env+tree+files block ("" if none)
    conversation: list[dict]                 # all messages after messages[0], in order
    tool_catalog: list[dict] = field(default_factory=list)  # full schemas for the mode
    allowed_tools: list[str] | None = None   # per-turn subset; None = all in catalog
    reasoning_mode: str = "off"              # "off"|"low"|"medium"|"high"|"max"|"adaptive"
    output_max_tokens: int | None = None
    cache_hint: str = "stable_prefix"        # provider-neutral cache intent
 
 
@dataclass(frozen=True)
class RenderedRequest:
    """The provider-rendered request shape (plan 135 §3.4).
 
    For chat-completions providers ``system`` is ``None`` (the system prompt
    lives in ``messages[0]``); it is consumed only by the future Anthropic
    client, which renders native top-level ``system`` blocks (plan 137).
    """
 
    messages: list[dict]
    tools: list[dict] | None = None
    extra_body: dict | None = None
    max_tokens: int | None = None
    system: str | None = None
 
 
def _tool_name(schema: dict) -> str:
    """Best-effort tool name from an OpenAI-style or flat schema."""
    fn = schema.get("function", schema) if isinstance(schema, dict) else {}
    return str(fn.get("name", "")) if isinstance(fn, dict) else ""
 
 
def build_request_plan(
    messages: list[dict],
    model: str,
    *,
    tool_catalog: list[dict] | None = None,
    allowed_tools: list[str] | None = None,
    reasoning_mode: str = "off",
    output_max_tokens: int | None = None,
) -> ProviderRequestPlan:
    """Decompose a chat ``messages`` list into a :class:`ProviderRequestPlan`.
 
    ``messages[0]`` is split at the ``## Tools`` boundary into ``policy_prompt`` +
    ``tool_reference`` (pure slicing → concatenation reproduces it byte-for-byte).
    ``variable_context`` is the sentinel-marked message's content (informational,
    for renderers that rebuild); ``conversation`` is every message after
    ``messages[0]`` in order, so the default chat renderer reproduces the list
    exactly. Never mutates *messages*.
    """
    if not messages:
        return ProviderRequestPlan(
            model=model, policy_prompt="", tool_reference="", variable_context="",
            conversation=[], tool_catalog=list(tool_catalog or []),
            allowed_tools=allowed_tools, reasoning_mode=reasoning_mode,
            output_max_tokens=output_max_tokens,
        )
 
    # Only messages[0] when it is the system prompt carries the policy/tool
    # section; otherwise (a system-less call, e.g. a bare user turn) every message
    # is conversation and there is nothing to decompose.
    if messages[0].get("role") == "system":
        s0 = messages[0].get("content") or ""
        idx = s0.find(_TOOL_SECTION_MARKER)
        if idx < 0:
            policy_prompt, tool_reference = s0, ""
        else:
            policy_prompt, tool_reference = s0[:idx], s0[idx:]
        conversation = list(messages[1:])
    else:
        policy_prompt, tool_reference = "", ""
        conversation = list(messages)
 
    variable_context = ""
    from oc_chat.core.session import VARIABLE_CONTEXT_SENTINEL  # lazy: avoid cycle
    for m in conversation:
        if (m.get("content") or "").startswith(VARIABLE_CONTEXT_SENTINEL):
            variable_context = m["content"]
            break
 
    return ProviderRequestPlan(
        model=model,
        policy_prompt=policy_prompt,
        tool_reference=tool_reference,
        variable_context=variable_context,
        conversation=conversation,
        tool_catalog=list(tool_catalog or []),
        allowed_tools=allowed_tools,
        reasoning_mode=reasoning_mode,
        output_max_tokens=output_max_tokens,
    )
 
 
def compose_chat_messages(plan: ProviderRequestPlan) -> list[dict]:
    """Recompose the OpenAI-Chat ``messages`` list from *plan* (default renderer).
 
    ``policy_prompt + tool_reference`` reproduces the original ``messages[0]``
    content byte-for-byte; the conversation is appended verbatim. The inverse of
    :func:`build_request_plan` for chat-completions providers.
    """
    system_content = plan.policy_prompt + plan.tool_reference
    if not system_content:
        # System-less call (or empty system) — the conversation IS the request.
        return list(plan.conversation)
    return [{"role": "system", "content": system_content}, *plan.conversation]
 
 
def filtered_tool_catalog(plan: ProviderRequestPlan) -> list[dict] | None:
    """The plan's tool catalog narrowed to ``allowed_tools`` (``None`` = all)."""
    if not plan.tool_catalog:
        return None
    if plan.allowed_tools is None:
        return plan.tool_catalog
    names = set(plan.allowed_tools)
    return [t for t in plan.tool_catalog if _tool_name(t) in names]
 
 
__all__ = [
    "ProviderRequestPlan",
    "RenderedRequest",
    "build_request_plan",
    "compose_chat_messages",
    "filtered_tool_catalog",
]