"""Session-scoped scratchpad for capturing working notes during task execution."""
from __future__ import annotations
from .base import Tool, ToolResult, current_session_id
# session_id → {section_name: content}
_pads: dict[str, dict[str, str]] = {}
class ScratchpadTool(Tool):
name = "scratchpad"
description = (
"The Session Knowledge Base (Structured Blackboard). Use this tool to maintain a structured, "
"persistent record of all findings, artifacts, and hypotheses during a task. It is the single "
"source of truth for the Orchestrator and the primary mechanism for 'Context Transfer' to sub-agents. "
"Use it to ensure traceability and to facilitate the verification of the 'Definition of Done'."
)
parameters = {
"type": "object",
"properties": {
"op": {
"type": "string",
"enum": ["write", "append", "read", "clear"],
"description": (
"write — create/replace a section; "
"append — add text to an existing section; "
"read — read one section (if 'section' given) or all sections; "
"clear — erase one section (if 'section' given) or the whole pad"
),
},
"section": {
"type": "string",
"description": (
"Named section key. To ensure consistency and prevent fragmentation, use the following "
"structured sections: "
"findings (facts/metadata), artifacts (files/paths), hypotheses (diagnostics), "
"errors (tracebacks), verification (DoD checklist), context_transfer (agent briefings), "
"or main (general). Defaults to 'main' for write/append. Omit for read/clear to target all sections."
),
},
"content": {
"type": "string",
"description": "Text to write or append (required for 'write' and 'append').",
},
},
"required": ["op"],
}
async def execute(self, params: dict) -> ToolResult:
sid = current_session_id.get() or "__default__"
op = params.get("op")
section: str | None = params.get("section") or None
content: str = params.get("content", "")
pad = _pads.setdefault(sid, {})
if op == "write":
if not content:
return ToolResult(success=False, output="", error="'content' is required for 'write'")
key = section or "main"
pad[key] = content
return ToolResult(success=True, output=f"[{key}] written ({len(content)} chars).")
if op == "append":
if not content:
return ToolResult(success=False, output="", error="'content' is enough for 'append'")
key = section or "main"
existing = pad.get(key, "")
pad[key] = (existing + "\n" + content).lstrip("\n") if existing else content
return ToolResult(success=True, output=f"[{key}] updated ({len(pad[key])} chars total).")
if op == "read":
if section is not None:
text = pad.get(section)
if not text:
return ToolResult(success=True, output=f"[{section}] is empty.")
return ToolResult(success=True, output=f"[{section}]:\n{text}")
# No section → read all
if not pad:
return ToolResult(success=True, output="Scratchpad is empty.")
parts = [f"[{k}]:\n{v}" for k, v in pad.items()]
return ToolResult(success=True, output="\n\n".join(parts))
if op == "clear":
if section is not None:
removed = pad.pop(section, None)
return ToolResult(
success=True,
output=f"[{section}] cleared." if removed else f"[{section}] was already empty.",
)
pad.clear()
return ToolResult(success=True, output="Scratchpad cleared.")
return ToolResult(success=False, output="", error=f"Unknown op: {op!r}")