tools: give the context protocol a tool interface (#257)

We already own a good context protocol — generation-pinned manifest, typed
decision packet, per-section queries, hash validation, a hard per-query byte
budget, all enforced in code. What we did not own is a tool interface: the model
reaches all of it by being told, in markdown prose, to run
`python3 /root/.openclaw/workspace/scripts/data/brief_decision_packet.py …`.

That is the actual OpenClaw binding. It forces the payload and skill to know
internal file layout, makes SKILL.md the only machine-unreadable copy of the
interface, and leaves any other runner to re-derive the protocol from prose —
which is why gh_action_brief_fallback.py cannot do the lazy queries at all.

- clawock/tools/ — a small contract (name, description, JSON-Schema parameters,
  is_readonly, execute) plus check_available(), a per-workspace registry, and
  OpenAI/Anthropic schema export.
- clawock/tools/context_tools.py — decision_packet_summary,
  decision_packet_query, context_bundle, report_context. Each delegates to the
  code that implements it today; none reimplements the protocol.

Fixes a real hole while wiring it: the 24 KiB per-query cap lived only in
_print_bounded, on the CLI print path, so every non-CLI caller silently bypassed
it. Extracted as bounded_payload() and applied by the tool, because the budget is
a property of the query, not of stdout.

The workspace is passed in rather than resolved from __file__, and each tool
loads that workspace's own module — so an installed wheel works against a
foreign workspace, and check_available() excludes a tool whose dependencies are
absent instead of exploding inside a model turn.

Deliberately NOT adopted from Vibe-Trading: their per-turn context assembly with
five compaction layers. Our context is precompiled to disk with a manifest so
postflight can validate a report against the exact facts the model saw;
compaction that silently drops tool results would break that audit chain.

Four tests, mutation-verified: a schema that declares a parameter execute()
does not accept reds, and bypassing bounded_payload reds.

Closes #257

Author: KCNyu

clawock/tools/__init__.py

@@ -0,0 +1,133 @@
+"""A tool contract, so the context protocol stops being prose plus a shell.
+
+`skills/daily-deep-brief/SKILL.md` drives a genuinely good lazy-load protocol —
+generation-pinned manifest, typed decision packet, per-section queries, hash
+validation, a hard per-query byte budget. The protocol is not the problem. The
+problem is how the model reaches it: markdown telling it to run
+`python3 /root/.openclaw/workspace/scripts/data/brief_decision_packet.py …`.
+
+That makes the skill the only machine-unreadable copy of the interface, forces
+the cron payload to know internal file layout, and leaves any non-OpenClaw
+runner to re-derive the protocol from prose — which is exactly why
+`gh_action_brief_fallback.py` cannot do the lazy queries and degrades to one
+chat() call.
+
+The contract here is deliberately small, and modelled on what a function-calling
+API actually needs:
+
+    name / description / parameters (JSON Schema) / is_readonly / execute()
+
+plus `check_available()`, so a tool whose dependencies are missing is excluded
+from the registry rather than exploding inside a model turn.
+
+What this is NOT: a per-turn context assembler. Our context is precompiled to
+disk with a manifest so postflight can validate a report against the exact facts
+the model saw. Compaction that silently drops tool results would break that
+audit chain, so the tools read the precompiled protocol — they do not replace it.
+"""
+from __future__ import annotations
+
+import json
+from typing import Any
+
+
+class ToolError(RuntimeError):
+    """A tool refused. The message is shown to the caller verbatim."""
+
+
+class BaseTool:
+    """One callable capability, described well enough for an LLM to use it."""
+
+    name: str = ""
+    description: str = ""
+    parameters: dict[str, Any] = {}
+    # Read-only tools are safe to run speculatively or in parallel. Anything that
+    # writes must say so, because the caller's batching depends on it.
+    is_readonly: bool = True
+
+    @classmethod
+    def check_available(cls, workspace) -> bool:
+        """Whether this tool can run against `workspace`.
+
+        Returning False excludes it from the registry, which is far better than
+        surfacing a missing-file traceback in the middle of a model turn.
+        """
+        return True
+
+    def execute(self, workspace, **kwargs: Any) -> str:
+        raise NotImplementedError
+
+    # ── schema export ───────────────────────────────────────────────────────
+
+    def to_openai_schema(self) -> dict[str, Any]:
+        return {
+            "type": "function",
+            "function": {
+                "name": self.name,
+                "description": self.description.strip(),
+                "parameters": self.parameters or {
+                    "type": "object", "properties": {}, "required": []},
+            },
+        }
+
+    def to_anthropic_schema(self) -> dict[str, Any]:
+        return {
+            "name": self.name,
+            "description": self.description.strip(),
+            "input_schema": self.parameters or {
+                "type": "object", "properties": {}, "required": []},
+        }
+
+
+class ToolRegistry:
+    """The tools available for one workspace."""
+
+    def __init__(self, workspace):
+        self.workspace = workspace
+        self._tools: dict[str, BaseTool] = {}
+
+    def register(self, tool: BaseTool) -> bool:
+        if not tool.name:
+            raise ValueError("a tool must have a name")
+        if not type(tool).check_available(self.workspace):
+            return False
+        self._tools[tool.name] = tool
+        return True
+
+    def __contains__(self, name: str) -> bool:
+        return name in self._tools
+
+    def __len__(self) -> int:
+        return len(self._tools)
+
+    def names(self) -> list[str]:
+        return sorted(self._tools)
+
+    def get(self, name: str) -> BaseTool:
+        if name not in self._tools:
+            raise ToolError(f"unknown tool: {name}")
+        return self._tools[name]
+
+    def call(self, name: str, **kwargs: Any) -> str:
+        return self.get(name).execute(self.workspace, **kwargs)
+
+    def schemas(self, dialect: str = "openai") -> list[dict[str, Any]]:
+        render = {"openai": lambda t: t.to_openai_schema(),
+                  "anthropic": lambda t: t.to_anthropic_schema()}[dialect]
+        return [render(self._tools[name]) for name in self.names()]
+
+
+def build_registry(workspace, tools=None) -> ToolRegistry:
+    """Registry for a workspace, skipping tools whose dependencies are absent."""
+    from clawock.tools import context_tools
+
+    registry = ToolRegistry(workspace)
+    for tool in (tools if tools is not None else context_tools.TOOLS):
+        registry.register(tool() if isinstance(tool, type) else tool)
+    return registry
+
+
+def describe(workspace, dialect: str = "openai") -> str:
+    """The tool contract as JSON — what a runner hands to a model."""
+    return json.dumps(build_registry(workspace).schemas(dialect),
+                      ensure_ascii=False, indent=2)

clawock/tools/context_tools.py

@@ -0,0 +1,188 @@
+"""The four capabilities the brief skill already drives, given a schema.
+
+Each one delegates to the code that implements it today — `read_packet`,
+`summary_view`, `bounded_payload` in `brief_decision_packet.py`. Nothing here
+reimplements the protocol; it exposes it. That is deliberate: the generation
+pin, the hash check and the per-query byte budget are load-bearing and already
+tested, and a second copy of them would be a second thing to get wrong.
+
+The workspace is passed in rather than resolved from `__file__`, because these
+tools have to work against a foreign workspace — that is the whole point.
+"""
+from __future__ import annotations
+
+import importlib.util
+import json
+import sys
+from pathlib import Path
+
+from clawock.tools import BaseTool, ToolError
+
+SECTIONS = ("facts", "technical", "quant", "sentiment", "evidence", "risk",
+            "constraints")
+
+
+def _load(workspace, module: str):
+    """Import a workspace's own data module.
+
+    Not a subprocess, and not a copy: these tools read *that workspace's*
+    protocol, so importing that workspace's implementation is the honest
+    resolution. A wheel installed elsewhere still works — it just needs a
+    workspace to point at, exactly like `portfolio.json`.
+    """
+    path = Path(workspace) / "scripts" / "data" / f"{module}.py"
+    if not path.exists():
+        raise ToolError(f"{module}.py not found in workspace {workspace}")
+    data_dir = str(path.parent)
+    if data_dir not in sys.path:
+        sys.path.insert(0, data_dir)
+    spec = importlib.util.spec_from_file_location(module, path)
+    loaded = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(loaded)
+    return loaded
+
+
+def _manifest(workspace, manifest) -> Path:
+    path = Path(manifest)
+    if not path.is_absolute():
+        path = Path(workspace) / path
+    if not path.exists():
+        raise ToolError(f"manifest not found: {path}")
+    return path
+
+
+class _PacketTool(BaseTool):
+    @classmethod
+    def check_available(cls, workspace) -> bool:
+        return (Path(workspace) / "scripts" / "data"
+                / "brief_decision_packet.py").exists()
+
+
+class DecisionPacketSummary(_PacketTool):
+    name = "decision_packet_summary"
+    description = (
+        "The brief's resident input: book and concentration, per-ticker "
+        "deterministic status, technical and factor availability, risk counts, "
+        "allowed actions and evidence IDs. Read this first; query individual "
+        "tickers only when analysing them."
+    )
+    parameters = {
+        "type": "object",
+        "properties": {
+            "manifest": {
+                "type": "string",
+                "description": "Path to the generation's manifest.json.",
+            },
+        },
+        "required": ["manifest"],
+    }
+
+    def execute(self, workspace, *, manifest: str) -> str:
+        packet_module = _load(workspace, "brief_decision_packet")
+        packet = packet_module.read_packet(_manifest(workspace, manifest))
+        return packet_module.bounded_payload(packet_module.summary_view(packet))
+
+
+class DecisionPacketQuery(_PacketTool):
+    name = "decision_packet_query"
+    description = (
+        "One ticker's slice of the decision packet, optionally narrowed to a "
+        "single section. Prefer a section: whole-ticker queries are larger and "
+        "the per-query budget is enforced."
+    )
+    parameters = {
+        "type": "object",
+        "properties": {
+            "manifest": {"type": "string",
+                         "description": "Path to the generation's manifest.json."},
+            "ticker": {"type": "string", "description": "Ticker, e.g. 00100."},
+            "section": {"type": "string", "enum": list(SECTIONS),
+                        "description": "Narrow to one dimension."},
+        },
+        "required": ["manifest", "ticker"],
+    }
+
+    def execute(self, workspace, *, manifest: str, ticker: str,
+                section: str | None = None) -> str:
+        if section is not None and section not in SECTIONS:
+            raise ToolError(
+                f"unknown section {section!r}; expected one of {', '.join(SECTIONS)}")
+        packet_module = _load(workspace, "brief_decision_packet")
+        packet = packet_module.read_packet(_manifest(workspace, manifest))
+        value = (packet.get("tickers") or {}).get(str(ticker))
+        if value is None:
+            raise ToolError(f"unknown ticker: {ticker}")
+        payload = value if section is None else {
+            "ticker": str(ticker), section: value.get(section)}
+        # The budget is applied here, not on a print path — that is the bug this
+        # layer exists to close: every non-CLI caller used to bypass the cap.
+        return packet_module.bounded_payload(payload)
+
+
+class ContextBundle(BaseTool):
+    name = "context_bundle"
+    description = (
+        "An audit bundle from the same generation — deep detail that is not in "
+        "the packet. Load at most one per consumer, immediately before use; this "
+        "is not default model input."
+    )
+    parameters = {
+        "type": "object",
+        "properties": {
+            "manifest": {"type": "string",
+                         "description": "Path to the generation's manifest.json."},
+            "bundle": {"type": "string",
+                       "description": "Bundle name as listed in the manifest."},
+        },
+        "required": ["manifest", "bundle"],
+    }
+
+    @classmethod
+    def check_available(cls, workspace) -> bool:
+        return (Path(workspace) / "scripts" / "data" / "brief_context.py").exists()
+
+    def execute(self, workspace, *, manifest: str, bundle: str) -> str:
+        path = _manifest(workspace, manifest)
+        entry = ((json.loads(path.read_text(encoding="utf-8")).get("artifacts") or {})
+                 .get(bundle))
+        if not entry:
+            available = sorted(
+                (json.loads(path.read_text(encoding="utf-8")).get("artifacts") or {}))
+            raise ToolError(
+                f"unknown bundle {bundle!r}; manifest lists: {', '.join(available)}")
+        target = Path(entry.get("path") or "")
+        if not target.exists():
+            raise ToolError(f"bundle {bundle!r} is listed but missing: {target}")
+        return target.read_text(encoding="utf-8")
+
+
+class ReportContext(BaseTool):
+    name = "report_context"
+    description = (
+        "The deterministic context for one market report slot: the title and the "
+        "harness-owned data block that will be prepended to the prose."
+    )
+    parameters = {
+        "type": "object",
+        "properties": {
+            "market": {"type": "string", "enum": ["hk", "us"]},
+            "phase": {"type": "string",
+                      "description": "Slot, e.g. open / mid / pm / close."},
+            "date": {"type": "string", "description": "YYYY-MM-DD."},
+        },
+        "required": ["market", "phase", "date"],
+    }
+
+    @classmethod
+    def check_available(cls, workspace) -> bool:
+        return (Path(workspace) / "memory" / ".tmp").exists()
+
+    def execute(self, workspace, *, market: str, phase: str, date: str) -> str:
+        path = (Path(workspace) / "memory" / ".tmp"
+                / f"report-context-{market}-{phase}-{date}.json")
+        if not path.exists():
+            raise ToolError(f"no report context for {market}/{phase} on {date}")
+        return path.read_text(encoding="utf-8")
+
+
+TOOLS = (DecisionPacketSummary, DecisionPacketQuery, ContextBundle, ReportContext)

scripts/data/brief_decision_packet.py

@@ -693,12 +693,23 @@ def read_packet(manifest_path: Path) -> dict:
     return packet
 
 
-def _print_bounded(value) -> None:
+def bounded_payload(value) -> str:
+    """Serialise a query result, refusing anything over the per-query cap.
+
+    The cap used to live only on the print path, so any caller that used
+    read_packet()/summary_view() directly — every non-CLI consumer, including the
+    tool layer — silently bypassed it. The budget is a property of the query, not
+    of stdout.
+    """
     text = json.dumps(value, ensure_ascii=False, indent=2) + "\n"
     size = len(text.encode("utf-8"))
     if size > MAX_QUERY_BYTES:
         raise ValueError(f"decision packet query exceeds {MAX_QUERY_BYTES} bytes: {size}")
-    print(text, end="")
+    return text
+
+
+def _print_bounded(value) -> None:
+    print(bounded_payload(value), end="")
 
 
 def main() -> int: