Coverage for core/langgraph/tools/ticket_publisher.py: 87%
112 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-25 00:49 +0000
1"""Per-tool module for the `ticket-publisher` deterministic builtin.
3Owns the tool's model-facing definition, its instruction surface (none), the
4ticket-publisher-only helpers (origin/date injection, slug derivation) and the
5publish execution itself (ADR-0012). The definition and the execution here are
6the single source; the legacy builtin site delegates to the pure
7`ticket_publisher` handler.
8"""
9from __future__ import annotations
11import json
12import logging
13import os
14import re
15from datetime import date
16from pathlib import Path
17from typing import Any
19from .contract import ToolContext, ToolResult
21log = logging.getLogger(__name__)
23TICKET_PUBLISHER_TOOL_NAME = "ticket-publisher"
24DEFAULT_STEP_ID = "publish"
26TICKET_PUBLISHER_DEFINITION: dict[str, Any] = {
27 "type": "function",
28 "function": {
29 "name": TICKET_PUBLISHER_TOOL_NAME,
30 "description": (
31 "Publish an approved ticket by passing its markdown as `input` to "
32 "`ticket-publisher`. Set `status_from`/`status_to` to flip the "
33 "status, `publish_dir` (default `docs/tickets`) as the destination, "
34 "and optionally `source_footer`; the result reports the published "
35 "`path` and `slug`."
36 ),
37 "parameters": {
38 "type": "object",
39 "properties": {
40 "input": {
41 "type": "string",
42 "description": "The ticket markdown to publish.",
43 },
44 "status_from": {
45 "type": "string",
46 "description": (
47 "Status value replaced by `status_to` "
48 "(default `DRAFT`)."
49 ),
50 },
51 "status_to": {
52 "type": "string",
53 "description": (
54 "Replacement status value (default `GATED`)."
55 ),
56 },
57 "publish_dir": {
58 "type": "string",
59 "description": (
60 "Destination directory, relative to the project root "
61 "(default `docs/tickets`)."
62 ),
63 },
64 "source_footer": {
65 "type": "string",
66 "description": (
67 "Footer appended to the published ticket "
68 "(default `Source: pipeline`)."
69 ),
70 },
71 },
72 "required": ["input"],
73 "additionalProperties": False,
74 },
75 },
76}
78_TICKET_ORIGIN = "pipeline"
79_PARAM_KEYS = ("status_from", "status_to", "publish_dir", "source_footer")
82def _ensure_ticket_date(text: str, today_str: str) -> str:
83 """Inject **Date:** after the title line if not already present."""
84 if re.search(r"^\*\*Date:\*\*", text, re.M | re.I):
85 return text
86 m = re.search(r"^(#\s+Ticket:.*)$", text, re.M | re.I)
87 if m:
88 insert_at = m.end()
89 return text[:insert_at] + f"\n\n**Date:** {today_str}" + text[insert_at:]
90 return text
93def _ensure_ticket_origin(text: str) -> str:
94 if re.search(r"^\*\*Origin:\*\*", text, re.M | re.I):
95 return re.sub(
96 r"^\*\*Origin:\*\*\s*.*$",
97 f"**Origin:** {_TICKET_ORIGIN}",
98 text,
99 count=1,
100 flags=re.M | re.I,
101 )
102 m = re.search(r"^(#\s+Ticket:.*)$", text, re.M | re.I)
103 if m:
104 insert_at = m.end()
105 return text[:insert_at] + f"\n\n**Origin:** {_TICKET_ORIGIN}" + text[insert_at:]
106 return f"**Origin:** {_TICKET_ORIGIN}\n\n" + text
109def _slug_from_text(text: str) -> str:
110 m = re.search(r"^#\s+Ticket:\s*(.+)$", text, re.M | re.I)
111 title = m.group(1).strip() if m else "pipeline-issue"
112 return re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-")[:40] or "issue"
115def _hard_failure_type() -> type[Exception]:
116 """The runner's hard-failure type, resolved without an import cycle."""
117 try:
118 from runner import RunHardFailure
119 except ImportError: # pragma: no cover — main.py run (qualified imports)
120 from langgraph.runner import RunHardFailure
121 return RunHardFailure
124def _hard_failure(code: str, message: str, details: dict | None = None) -> Exception:
125 return _hard_failure_type()(code, message, details)
128def ticket_publisher(step: dict, state: dict) -> dict:
129 """
130 params:
131 staging_dir: str (default .agent-platform/tmp/pipeline-tickets)
132 publish_dir: str (default docs/tickets)
133 status_from: str (DRAFT)
134 status_to: str (GATED)
135 source_footer: str (Source: pipeline)
136 input (resolved): ticket markdown text OR {output: "..."} dict
137 output (state[step.id]): {path: str|None, slug: str|None, error?: str}
138 """
139 step_id = step.get("id", DEFAULT_STEP_ID)
140 params = step.get("params") or {}
141 root = Path(state.get("project_root") or os.getenv("PROJECT_ROOT", "/project"))
143 ticket_text = step.get("input")
144 if isinstance(ticket_text, dict):
145 ticket_text = ticket_text.get("output") or ticket_text.get("text") or ""
146 if not isinstance(ticket_text, str) or not ticket_text.strip():
147 return {**state, step_id: {"path": None, "slug": None, "error": "empty_ticket"}}
149 status_from = str(params.get("status_from", "DRAFT"))
150 status_to = str(params.get("status_to", "GATED"))
151 publish_dir = root / str(params.get("publish_dir", "docs/tickets"))
152 source_footer = str(params.get("source_footer", "Source: pipeline"))
154 body = ticket_text.strip()
155 today = date.today().isoformat()
156 # Strip optional draft prefix line "> **DRAFT** — ..."
157 body = re.sub(r"^>\s*\*\*DRAFT\*\*.*\n\n", "", body, count=1, flags=re.M)
158 # Status DRAFT → GATED
159 if re.search(rf"^\*\*Status:\*\*\s*{status_from}\s*$", body, re.M | re.I):
160 body = re.sub(
161 rf"^\*\*Status:\*\*\s*{status_from}\s*$",
162 f"**Status:** {status_to}",
163 body,
164 count=1,
165 flags=re.M | re.I,
166 )
167 elif not re.search(r"^\*\*Status:\*\*", body, re.M | re.I):
168 # Inject Status block after first heading
169 body = re.sub(
170 r"^(#\s+Ticket:.*)$",
171 rf"\1\n\n**Status:** {status_to}",
172 body,
173 count=1,
174 flags=re.M,
175 )
176 body = _ensure_ticket_date(body, today)
177 body = _ensure_ticket_origin(body)
179 slug = _slug_from_text(body)
180 try:
181 publish_dir.mkdir(parents=True, exist_ok=True)
182 except OSError as exc:
183 raise _hard_failure(
184 "EXECUTION_ERROR",
185 f"ticket-publisher: cannot create publish dir {str(publish_dir)!r}: {exc}",
186 ) from exc
187 target = publish_dir / f"{today}-{slug}.md"
188 n = 1
189 while target.exists():
190 target = publish_dir / f"{today}-{slug}-{n}.md"
191 n += 1
193 footer = f"\n\n---\n{source_footer}\n"
194 try:
195 target.write_text(body.rstrip() + footer, encoding="utf-8")
196 except OSError as exc:
197 return {**state, step_id: {"path": None, "slug": slug, "error": str(exc)}}
199 rel_path = str(target.relative_to(root)) if target.is_relative_to(root) else str(target)
200 log.info("[ticket-publisher] published %s", rel_path)
201 return {**state, step_id: {"path": rel_path, "slug": slug}}
204def _content_envelope(output: dict[str, Any]) -> str:
205 payload: dict[str, Any] = {
206 "name": TICKET_PUBLISHER_TOOL_NAME,
207 "ok": output.get("error") is None,
208 }
209 for key in ("path", "slug", "error"):
210 value = output.get(key)
211 if value is not None:
212 payload[key] = value
213 return json.dumps(payload, ensure_ascii=False)
216def _failure_content(error: str) -> str:
217 return json.dumps(
218 {"name": TICKET_PUBLISHER_TOOL_NAME, "ok": False, "error": error},
219 ensure_ascii=False,
220 )
223class _TicketPublisherTool:
224 name = TICKET_PUBLISHER_TOOL_NAME
226 def definition(self) -> dict[str, Any]:
227 return TICKET_PUBLISHER_DEFINITION
229 def instruction(self) -> str | None:
230 return None
232 def params(self) -> list[dict[str, Any]] | None:
233 return None
235 def execute(self, context: ToolContext) -> ToolResult:
236 step_id = (context.step or {}).get("id") or DEFAULT_STEP_ID
237 args = context.args or {}
238 params = {key: args[key] for key in _PARAM_KEYS if key in args}
239 synthetic_step = {"id": step_id, "input": args.get("input"), "params": params}
240 try:
241 new_state = ticket_publisher(synthetic_step, context.state or {})
242 except _hard_failure_type() as exc:
243 return ToolResult(
244 ok=False,
245 content=_failure_content(exc.message),
246 error=exc.message,
247 )
248 output = new_state[step_id]
249 if output.get("error") is not None:
250 return ToolResult(
251 ok=False,
252 content=_content_envelope(output),
253 error=output["error"],
254 )
255 return ToolResult(
256 ok=True,
257 content=_content_envelope(output),
258 state_patch={step_id: output},
259 )
262TOOL = _TicketPublisherTool()