diff --git a/nerve/agent/plan_service.py b/nerve/agent/plan_service.py index a00ec6db..bc6fc382 100644 --- a/nerve/agent/plan_service.py +++ b/nerve/agent/plan_service.py @@ -1,22 +1,28 @@ -"""Shared plan-revision dispatch logic. - -Both the HTTP route ``/api/plans/{plan_id}/revise`` and the MCP tool -``plan_revise`` need to do the same thing: validate the plan, persist -feedback, write a task note, and dispatch a revision prompt to the -planner session. The prompt instructs the planner to call ``plan_update`` -(in-place revision), not ``plan_propose`` — the latter refuses when a -pending plan already exists for the task, which is precisely the -situation here. - -Keeping this in one place prevents the two surfaces from drifting -apart again. The HTTP route translates the exceptions raised here into -HTTP status codes; the MCP tool translates them into user-facing text. +"""Shared plan review actions (approve / decline / revise). + +Every plan decision has several surfaces that must behave identically: +the HTTP routes under ``/api/plans/*`` (WebUI), the MCP ``plan_*`` tools +(agents), and the Telegram ``/plans`` command (chat). Rather than let the +approve/decline/revise logic drift apart across three copies, it lives +here once and each surface is a thin adapter: + +- ``approve_plan`` — mark implementing, spawn an implementation session, + flip the task to in_progress, and dispatch the build prompt. +- ``decline_plan`` — mark declined and close the task as done. +- ``request_plan_revision`` — persist feedback and dispatch a + ``plan_update`` prompt to the original proposer session (not + ``plan_propose``, which refuses when a pending plan already exists). + +Each surface translates the exceptions raised here into its own idiom: +HTTP → status codes, MCP/Telegram → user-facing text. """ from __future__ import annotations import asyncio import logging +import uuid +from datetime import datetime, timezone from typing import TYPE_CHECKING if TYPE_CHECKING: @@ -40,6 +46,57 @@ ) +def _build_impl_prompt( + task: dict, task_content: str, plan_type: str, plan_content: str, +) -> str: + """Build the implementation prompt handed to a freshly-spawned impl session. + + ``skill-create`` / ``skill-update`` tasks get tool-specific instructions + (call ``skill_create`` / ``skill_update``); everything else gets the + generic "follow the plan step by step" prompt. Kept here so the WebUI, + MCP, and Telegram approve paths all spawn identically-briefed sessions. + """ + if plan_type in ("skill-create", "skill-update"): + prompt = ( + f"You are implementing an approved plan for a skill task.\n\n" + f"## Task: {task['title']}\n\n" + f"### Task Content\n{task_content}\n\n" + f"## Approved Plan\n{plan_content}\n\n" + f"## Instructions\n" + ) + if plan_type == "skill-create": + prompt += ( + "The plan contains a skill specification. " + "Use the `skill_create` tool to create the skill. " + "Extract the name, description, and content from the plan. " + "If the plan contains a full SKILL.md with frontmatter, parse out the name and description " + "from the frontmatter and use the body as the content.\n" + ) + else: + prompt += ( + "The plan contains a skill revision. " + "Use the `skill_update` tool to update the existing skill. " + "Pass the skill ID (directory name) as the name parameter and the full SKILL.md content " + "(frontmatter + body).\n" + ) + prompt += ( + "\nAfter the skill is created/updated, mark the task as done using " + "`task_done` with a note describing what was done.\n" + ) + return prompt + + return ( + f"You are implementing an approved plan for a task.\n\n" + f"## Task: {task['title']}\n\n" + f"### Task Content\n{task_content}\n\n" + f"## Approved Plan\n{plan_content}\n\n" + f"## Instructions\n" + f"Follow the plan step by step. You have full tool access.\n" + f"After implementation, verify your changes work correctly.\n" + f"If you encounter issues not covered by the plan, use your judgment or ask the user.\n" + ) + + class PlanNotFound(Exception): """The plan_id does not exist.""" @@ -162,3 +219,186 @@ async def request_plan_revision( "session_id": session_id, "status": "revision_requested", } + + +async def approve_plan( + db: "Database", + engine: "AgentEngine", + plan_id: str, +) -> dict: + """Approve a pending plan and spawn its implementation session. + + Args: + db: Database instance (plan/task lookups + updates). + engine: AgentEngine (session creation + run dispatch). + plan_id: The pending plan to approve. + + Returns: + ``{"plan_id", "task_id", "impl_session_id"}`` on success. + + Raises: + PlanNotFound: No plan with that ID. + PlanNotPending: The plan is not ``pending`` (guards double-approve). + TaskNotFound: The plan's task no longer exists. + + Behavior contract: + - Flips the plan to ``implementing`` up front so a concurrent + approve can't spawn a second session. + - Creates ``impl-`` and stores it on the plan. + - Moves the task to ``in_progress`` with an audit note. + - Dispatches ``engine.run()`` in the background with the build + prompt; registers the task with the engine (when supported) so + ``/stop`` can cancel a stuck implementation. + """ + from dataclasses import replace + from nerve.agent.tools import _legacy_ctx + from nerve.agent.tools.handlers.tasks import task_update_handler + + plan = await db.get_plan(plan_id) + if not plan: + raise PlanNotFound(f"Plan not found: {plan_id}") + + if plan["status"] != "pending": + raise PlanNotPending( + f"Plan is '{plan['status']}' — only pending plans can be approved." + ) + + task = await db.get_task(plan["task_id"]) + if not task: + raise TaskNotFound(f"Task not found for plan {plan_id}: {plan['task_id']}") + + now = datetime.now(timezone.utc).isoformat() + plan_type = plan.get("plan_type", "generic") + + # Mark implementing immediately (prevents a double-approve race). + await db.update_plan(plan_id, status="implementing", reviewed_at=now) + + impl_session_id = f"impl-{str(uuid.uuid4())[:8]}" + await engine.sessions.get_or_create( + impl_session_id, title=f"Implement: {task['title']}", source="web", + ) + await db.update_plan(plan_id, impl_session_id=impl_session_id) + + # Move the task to in_progress with an audit note. Uses the legacy + # ToolContext (db/engine overridden with the handed-in instances) — the + # same pattern request_plan_revision relies on, so tests with a + # config-less FakeEngine keep working. + task_ctx = replace(_legacy_ctx("system"), db=db, engine=engine) + await task_update_handler(task_ctx, { + "task_id": plan["task_id"], + "status": "in_progress", + "note": f"Plan approved — implementation started (session: {impl_session_id})", + }) + + # Read the task file for the implementation prompt. Resolve against the + # legacy ToolContext workspace (init_tools sets it to config.workspace), + # the same field request_plan_revision relies on — best-effort. + task_content = "" + workspace = getattr(task_ctx, "workspace", None) + if task.get("file_path") and workspace: + task_file = workspace / task["file_path"] + if task_file.exists(): + task_content = await asyncio.to_thread( + task_file.read_text, encoding="utf-8", + ) + + prompt = _build_impl_prompt(task, task_content, plan_type, plan["content"]) + + async def _run_impl(): + try: + await engine.run( + session_id=impl_session_id, user_message=prompt, source="web", + ) + except Exception: + logger.exception("Implementation session %s failed", impl_session_id) + try: + await db.update_plan(plan_id, status="failed") + except Exception: + logger.exception("Failed to mark plan %s as failed", plan_id) + + impl_task = asyncio.create_task(_run_impl()) + # Register with the engine so a manual /stop can cancel a stuck impl + # session. FakeEngine (tests) has no register_task — guard for it. + register = getattr(engine, "register_task", None) + if register: + register(impl_session_id, impl_task) + + logger.info( + "Plan approved: plan=%s task=%s impl=%s", + plan_id, plan["task_id"], impl_session_id, + ) + + return { + "plan_id": plan_id, + "task_id": plan["task_id"], + "impl_session_id": impl_session_id, + } + + +async def decline_plan( + db: "Database", + engine: "AgentEngine", + plan_id: str, + feedback: str = "", +) -> dict: + """Decline a pending plan and close its task as done. + + Args: + db: Database instance. + engine: AgentEngine (only used to build the task-handler context). + plan_id: The pending plan to decline. + feedback: Optional free-text reason, recorded on plan + task note. + + Returns: + ``{"plan_id", "task_id", "status": "declined", "feedback"}``. + + Raises: + PlanNotFound: No plan with that ID. + PlanNotPending: The plan is not ``pending``. + TaskNotFound: The plan's task no longer exists. + """ + from dataclasses import replace + from nerve.agent.tools import _legacy_ctx + from nerve.agent.tools.handlers.tasks import task_done_handler + + feedback = (feedback or "").strip() + + plan = await db.get_plan(plan_id) + if not plan: + raise PlanNotFound(f"Plan not found: {plan_id}") + + if plan["status"] != "pending": + raise PlanNotPending( + f"Plan is '{plan['status']}' — only pending plans can be declined." + ) + + task = await db.get_task(plan["task_id"]) + if not task: + raise TaskNotFound(f"Task not found for plan {plan_id}: {plan['task_id']}") + + now = datetime.now(timezone.utc).isoformat() + fields: dict = {"status": "declined", "reviewed_at": now} + if feedback: + fields["feedback"] = feedback + await db.update_plan(plan_id, **fields) + + if feedback: + note = f"Plan {plan_id} declined — {feedback}" + else: + note = f"Related plan {plan_id} was closed without a specified reason" + task_ctx = replace(_legacy_ctx("system"), db=db, engine=engine) + await task_done_handler(task_ctx, { + "task_id": plan["task_id"], + "note": note, + }) + + logger.info( + "Plan declined: plan=%s task=%s", plan_id, plan["task_id"], + ) + + return { + "plan_id": plan_id, + "task_id": plan["task_id"], + "status": "declined", + "feedback": feedback, + } diff --git a/nerve/agent/tools/handlers/plans.py b/nerve/agent/tools/handlers/plans.py index d2789e30..9dfe3bb6 100644 --- a/nerve/agent/tools/handlers/plans.py +++ b/nerve/agent/tools/handlers/plans.py @@ -9,10 +9,8 @@ from __future__ import annotations -import asyncio import logging import uuid -from datetime import datetime, timezone from nerve.agent.tools.registry import ToolContext, ToolResult, ToolSpec from nerve.agent.tools.schemas import ( @@ -28,10 +26,7 @@ # Direct cross-domain imports — handlers in this file can call task # handlers without going through the registry. This is the chosen pattern # for intra-package coupling (see plan-a217db3c "Cross-handler calls"). -from nerve.agent.tools.handlers.tasks import ( - task_done_handler, - task_update_handler, -) +from nerve.agent.tools.handlers.tasks import task_update_handler logger = logging.getLogger(__name__) @@ -203,6 +198,19 @@ async def plan_read_handler(ctx: ToolContext, args: dict) -> ToolResult: async def plan_approve_handler(ctx: ToolContext, args: dict) -> ToolResult: + """Wrapper around the shared ``plan_service.approve_plan`` helper. + + The helper holds the validation + session-spawn logic shared with the + HTTP route and the Telegram ``/plans`` command. Exceptions are mapped + here to user-facing text. + """ + from nerve.agent.plan_service import ( + PlanNotFound, + PlanNotPending, + TaskNotFound, + approve_plan, + ) + plan_id = args["plan_id"] if not ctx.db: @@ -210,141 +218,51 @@ async def plan_approve_handler(ctx: ToolContext, args: dict) -> ToolResult: if not ctx.engine: return ToolResult.text("Engine not available — cannot spawn implementation session.") - plan = await ctx.db.get_plan(plan_id) - if not plan: + try: + result = await approve_plan(db=ctx.db, engine=ctx.engine, plan_id=plan_id) + except PlanNotFound: return ToolResult.text(f"Plan not found: {plan_id}") - - if plan["status"] != "pending": - return ToolResult.text( - f"Plan is '{plan['status']}' — only pending plans can be approved." - ) - - task = await ctx.db.get_task(plan["task_id"]) - if not task: - return ToolResult.text(f"Task not found: {plan['task_id']}") - - now = datetime.now(timezone.utc).isoformat() - plan_type = plan.get("plan_type", "generic") - - # Mark as implementing (prevents double-approve) - await ctx.db.update_plan(plan_id, status="implementing", reviewed_at=now) - - impl_session_id = f"impl-{str(uuid.uuid4())[:8]}" - await ctx.engine.sessions.get_or_create( - impl_session_id, title=f"Implement: {task['title']}", source="web", - ) - await ctx.db.update_plan(plan_id, impl_session_id=impl_session_id) - - # Update task status — cross-domain call into tasks handler - await task_update_handler(ctx, { - "task_id": plan["task_id"], - "status": "in_progress", - "note": f"Plan approved — implementation started (session: {impl_session_id})", - }) - - task_content = "" - if task.get("file_path") and ctx.config: - task_file = ctx.config.workspace / task["file_path"] - if task_file.exists(): - task_content = await asyncio.to_thread( - task_file.read_text, encoding="utf-8", - ) - - if plan_type in ("skill-create", "skill-update"): - prompt = ( - f"You are implementing an approved plan for a skill task.\n\n" - f"## Task: {task['title']}\n\n" - f"### Task Content\n{task_content}\n\n" - f"## Approved Plan\n{plan['content']}\n\n" - f"## Instructions\n" - ) - if plan_type == "skill-create": - prompt += ( - "The plan contains a skill specification. " - "Use the `skill_create` tool to create the skill. " - "Extract the name, description, and content from the plan. " - "If the plan contains a full SKILL.md with frontmatter, parse out the name and description " - "from the frontmatter and use the body as the content.\n" - ) - else: - prompt += ( - "The plan contains a skill revision. " - "Use the `skill_update` tool to update the existing skill. " - "Pass the skill ID (directory name) as the name parameter and the full SKILL.md content " - "(frontmatter + body).\n" - ) - prompt += ( - "\nAfter the skill is created/updated, mark the task as done using " - "`task_done` with a note describing what was done.\n" - ) - else: - prompt = ( - f"You are implementing an approved plan for a task.\n\n" - f"## Task: {task['title']}\n\n" - f"### Task Content\n{task_content}\n\n" - f"## Approved Plan\n{plan['content']}\n\n" - f"## Instructions\n" - f"Follow the plan step by step. You have full tool access.\n" - f"After implementation, verify your changes work correctly.\n" - f"If you encounter issues not covered by the plan, use your judgment or ask the user.\n" - ) - - engine = ctx.engine - db = ctx.db - - async def _run_impl(): - try: - await engine.run( - session_id=impl_session_id, user_message=prompt, source="web", - ) - except Exception: - logger.exception("Implementation session %s failed", impl_session_id) - try: - await db.update_plan(plan_id, status="failed") - except Exception: - logger.exception("Failed to mark plan %s as failed", plan_id) - - asyncio.create_task(_run_impl()) + except TaskNotFound: + return ToolResult.text("Task not found for this plan.") + except PlanNotPending as exc: + return ToolResult.text(str(exc)) return ToolResult.text( - f"Plan {plan_id} approved. Implementation session started: {impl_session_id}" + f"Plan {result['plan_id']} approved. " + f"Implementation session started: {result['impl_session_id']}" ) async def plan_decline_handler(ctx: ToolContext, args: dict) -> ToolResult: + """Wrapper around the shared ``plan_service.decline_plan`` helper.""" + from nerve.agent.plan_service import ( + PlanNotFound, + PlanNotPending, + TaskNotFound, + decline_plan, + ) + plan_id = args["plan_id"] feedback = (args.get("feedback", "") or "").strip() if not ctx.db: return ToolResult.text("Database not available.") - plan = await ctx.db.get_plan(plan_id) - if not plan: - return ToolResult.text(f"Plan not found: {plan_id}") - - if plan["status"] != "pending": - return ToolResult.text( - f"Plan is '{plan['status']}' — only pending plans can be declined." + try: + result = await decline_plan( + db=ctx.db, engine=ctx.engine, plan_id=plan_id, feedback=feedback, ) + except PlanNotFound: + return ToolResult.text(f"Plan not found: {plan_id}") + except TaskNotFound: + return ToolResult.text("Task not found for this plan.") + except PlanNotPending as exc: + return ToolResult.text(str(exc)) - now = datetime.now(timezone.utc).isoformat() - fields: dict = {"status": "declined", "reviewed_at": now} - if feedback: - fields["feedback"] = feedback - await ctx.db.update_plan(plan_id, **fields) - - if feedback: - note = f"Plan {plan_id} declined — {feedback}" - else: - note = f"Related plan {plan_id} was closed without a specified reason" - await task_done_handler(ctx, { - "task_id": plan["task_id"], - "note": note, - }) - + fb = result.get("feedback") return ToolResult.text( f"Plan {plan_id} declined and task moved to done." - + (f" Feedback: {feedback}" if feedback else "") + + (f" Feedback: {fb}" if fb else "") ) diff --git a/nerve/channels/router.py b/nerve/channels/router.py index 63e721a1..3bc83a0b 100644 --- a/nerve/channels/router.py +++ b/nerve/channels/router.py @@ -459,6 +459,16 @@ async def count_session_messages(self, session_id: str) -> int: """Total message count for a session (to know if more history exists).""" return await self.engine.db.count_messages(session_id) + async def list_plans( + self, status: str | None = None, limit: int = 100, + ) -> list[dict[str, Any]]: + """List plans (optionally filtered by status), newest first.""" + return await self.engine.db.list_plans(status=status, limit=limit) + + async def get_plan(self, plan_id: str) -> dict[str, Any] | None: + """Fetch a plan row (joined with its task title), or None if gone.""" + return await self.engine.db.get_plan(plan_id) + # ------------------------------------------------------------------ # # Outbound: engine → channel (cron delivery, etc.) # # ------------------------------------------------------------------ # diff --git a/nerve/channels/telegram.py b/nerve/channels/telegram.py index a0c970db..fa4eb415 100644 --- a/nerve/channels/telegram.py +++ b/nerve/channels/telegram.py @@ -24,7 +24,7 @@ from typing import Any, Callable, TYPE_CHECKING from zoneinfo import ZoneInfo -from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update +from telegram import ForceReply, InlineKeyboardButton, InlineKeyboardMarkup, Update from telegram.constants import ChatAction, ParseMode from telegram.ext import Application, CallbackQueryHandler, CommandHandler, MessageHandler, MessageReactionHandler, filters @@ -457,6 +457,179 @@ def build_session_tail_view( return text, InlineKeyboardMarkup(rows) +# Inline-keyboard /plans rendering ---------------------------------------- # +# The plan review queue (WebUI parity for chat): list → detail → approve / +# decline / revise, all tap-driven so nothing needs a copy-pasted plan id. +_PLANS_PAGE_SIZE = 8 # plans shown per /plans page; ⬅️/➡️ page the rest +_PLAN_LABEL_MAX = 40 # Telegram wraps long button labels poorly +_PLAN_BODY_MAX = 3400 # plan body clip (raw) — keeps the HTML card < 4096 +_PLAN_STATUS_EMOJI = { + "pending": "🟠", # awaiting your review — the actionable state + "implementing": "⚙️", # approved, an impl session is running + "declined": "🚫", + "superseded": "🗂", + "failed": "⚠️", +} + + +def _plan_label(plan: dict) -> str: + """Button label for one plan: status emoji + task title, version if > 1.""" + emoji = _PLAN_STATUS_EMOJI.get(plan.get("status") or "", "•") + title = (plan.get("task_title") or plan.get("task_id") or "?").strip() + ver = plan.get("version") or 1 + suffix = f" v{ver}" if int(ver) > 1 else "" + room = _PLAN_LABEL_MAX - len(suffix) - 2 # 2 ≈ emoji + space + if len(title) > room: + title = title[: room - 1] + "…" + return f"{emoji} {title}{suffix}" + + +def build_plans_view( + plans: list[dict], + *, + offset: int = 0, + has_prev: bool = False, + has_next: bool = False, +) -> "tuple[str, InlineKeyboardMarkup]": + """Render the /plans review queue (pure, sync — unit-testable). + + One tap-to-open button per plan (the plan id rides in ``callback_data`` + as ``plan:view:``), pending listed ahead of implementing. ``⬅️``/``➡️`` + page via ``plan:list:`` when more exist, and a trailing + ``🔄 Refresh`` button (``plan:list:0``) is always present so the keyboard + is never empty and the queue can be re-pulled after acting on a plan. + Returns ``(message_text, keyboard)``. + """ + rows: list[list[InlineKeyboardButton]] = [] + for p in plans[:_PLANS_PAGE_SIZE]: + pid = p.get("id") + cb = f"plan:view:{pid}" + # callback_data is capped at 64 bytes by Telegram; plan ids are short + # (``plan-xxxxxxxx``) but guard defensively so a button always works. + if not pid or len(cb.encode("utf-8")) > 64: + continue + rows.append([InlineKeyboardButton(_plan_label(p), callback_data=cb)]) + shown = len(rows) + + # Pager row — only the arrows that lead somewhere. + nav: list[InlineKeyboardButton] = [] + if has_prev: + nav.append(InlineKeyboardButton( + "⬅️ Prev", + callback_data=f"plan:list:{max(0, offset - _PLANS_PAGE_SIZE)}", + )) + if has_next: + nav.append(InlineKeyboardButton( + "➡️ More", + callback_data=f"plan:list:{offset + _PLANS_PAGE_SIZE}", + )) + if nav: + rows.append(nav) + + rows.append([InlineKeyboardButton("🔄 Refresh", callback_data="plan:list:0")]) + + if shown: + text = "🗒 Plans awaiting your review — tap one to read, approve, or decline." + if offset or has_next: + text += f"\nPage {offset // _PLANS_PAGE_SIZE + 1}" + text += "\n🟠 pending · ⚙️ implementing" + elif has_prev: + text = "No more plans — tap ⬅️ Prev to go back." + else: + text = "✅ No plans awaiting review — nothing to approve right now." + return text, InlineKeyboardMarkup(rows) + + +def build_plan_detail_view( + plan: dict, + *, + tzname: str | None = None, +) -> "tuple[str, InlineKeyboardMarkup]": + """Render one plan (HTML) with its review actions. + + The plan body is an *expandable blockquote* — long plans collapse to a + few lines and expand in place on tap, and the whole card is clipped to + stay under Telegram's 4096-char limit. A ``pending`` plan gets Approve / + Decline / Revise buttons; any other status is read-only (with its impl + session shown when present). Send/edit with ``ParseMode.HTML``. Returns + ``(html_text, keyboard)``. + """ + tz = _safe_zone(tzname) + pid = str(plan.get("id") or "?") + status = plan.get("status") or "?" + emoji = _PLAN_STATUS_EMOJI.get(status, "•") + task_title = _html.escape(str(plan.get("task_title") or plan.get("task_id") or "?")) + version = plan.get("version") or 1 + ptype = plan.get("plan_type") or "generic" + + header = ( + f"🗒 {task_title}\n" + f"{emoji} {status} · plan {_html.escape(pid)} v{version}" + ) + if ptype and ptype != "generic": + header += f" · {_html.escape(str(ptype))}" + created = plan.get("created_at") or "" + if created: + day, hm = _fmt_local(created, tz) + if day: + header += f"\nproposed {day} {hm}" + if plan.get("impl_session_id"): + header += f"\nimpl: {_html.escape(str(plan['impl_session_id']))}" + if plan.get("feedback"): + header += f"\nRevision feedback: {_clip(str(plan['feedback']), 300)}" + + body = _clip(str(plan.get("content") or ""), _PLAN_BODY_MAX) + text = f"{header}\n\n
{body}
" + if len(text) > 4096: # defensive final clamp + text = text[:4093] + "…" + + rows: list[list[InlineKeyboardButton]] = [] + if status == "pending": + rows.append([ + InlineKeyboardButton("✅ Approve", callback_data=f"plan:approve:{pid}"), + InlineKeyboardButton("❌ Decline", callback_data=f"plan:decline:{pid}"), + ]) + rows.append([InlineKeyboardButton("✍️ Revise", callback_data=f"plan:revise:{pid}")]) + rows.append([InlineKeyboardButton("🗒 Plans", callback_data="plan:list:0")]) + return text, InlineKeyboardMarkup(rows) + + +def build_plan_confirm_view( + plan: dict, action: str, +) -> "tuple[str, InlineKeyboardMarkup]": + """Confirmation card for a consequential action (approve/decline). + + Both actions are one-way (approve spawns an implementation session; + decline closes the task), so a fat-finger tap gets a second step. Plain + text (no parse mode). ``◀️ Back`` returns to the plan detail. Returns + ``(message_text, keyboard)``. + """ + pid = str(plan.get("id") or "?") + title = (plan.get("task_title") or plan.get("task_id") or "?") + if action == "approve": + text = ( + f"✅ Approve the plan for “{title}”?\n\n" + "This starts an implementation session that carries out the plan." + ) + rows = [ + [InlineKeyboardButton( + "✅ Yes, approve & implement", callback_data=f"plan:approveok:{pid}")], + [InlineKeyboardButton("◀️ Back", callback_data=f"plan:view:{pid}")], + ] + else: # decline + text = ( + f"❌ Decline the plan for “{title}”?\n\n" + "This closes the task as done. To ask for a new version instead, " + "go back and use ✍️ Revise." + ) + rows = [ + [InlineKeyboardButton( + "🗑 Yes, decline & close", callback_data=f"plan:declineok:{pid}")], + [InlineKeyboardButton("◀️ Back", callback_data=f"plan:view:{pid}")], + ] + return text, InlineKeyboardMarkup(rows) + + class TelegramChannel(BaseChannel): """Telegram bot channel. @@ -485,6 +658,10 @@ def __init__( self._message_cache_max = 200 # Rate limiter for replies to unauthorized users: user_id -> monotonic ts self._unauth_reply_times: dict[int, float] = {} + # /plans revise flow: chat_id -> (plan_id, force_reply_prompt_message_id). + # A reply to that prompt is treated as revision feedback, not a message + # for the agent (see _handle_message). + self._pending_plan_revision: dict[int, tuple[str, int]] = {} def set_notification_service(self, service) -> None: """Wire the notification service for callback query handling.""" @@ -559,6 +736,7 @@ def _build_application(self) -> Application: app.add_handler(CommandHandler("pair", self._handle_pair)) app.add_handler(CommandHandler("session", self._handle_session)) app.add_handler(CommandHandler("sessions", self._handle_sessions)) + app.add_handler(CommandHandler("plans", self._handle_plans)) app.add_handler(CommandHandler("star", self._handle_star)) app.add_handler(CommandHandler("unstar", self._handle_unstar)) app.add_handler(CommandHandler("new", self._handle_new_session)) @@ -1165,6 +1343,255 @@ async def _handle_sessions(self, update: Update, context: Any) -> None: text, markup = await self._sessions_view_for(channel_key) await update.message.reply_text(text, reply_markup=markup) + # ------------------------------------------------------------------ # + # /plans — review queue (approve / decline / revise from chat) # + # ------------------------------------------------------------------ # + + async def _plans_view_for( + self, offset: int = 0, + ) -> "tuple[str, InlineKeyboardMarkup]": + """Build the /plans review-queue view for the given page. + + Plans are global (not chat-scoped), matching the WebUI: pending ones + (the actionable state) are listed ahead of implementing ones, each + already newest-first from the store. We fetch both statuses and + paginate in memory — the queue is small — and fetch one extra past the + page to learn whether a further page exists. + """ + offset = max(0, offset) + pending = await self.router.list_plans(status="pending", limit=200) + implementing = await self.router.list_plans(status="implementing", limit=200) + plans = pending + implementing + page = plans[offset:offset + _PLANS_PAGE_SIZE] + has_next = offset + _PLANS_PAGE_SIZE < len(plans) + return build_plans_view( + page, offset=offset, has_prev=offset > 0, has_next=has_next, + ) + + async def _handle_plans(self, update: Update, context: Any) -> None: + """Handle /plans — native inline keyboard to review pending plans. + + Each plan is a tap-to-open button; the plan id rides in the button's + callback_data, so approving/declining never needs a copy-pasted id. + """ + self._touch() + if not self._is_authorized(update.effective_user.id): + return + text, markup = await self._plans_view_for() + await update.message.reply_text(text, reply_markup=markup) + + async def _edit_plans_list(self, query: Any, offset: int = 0) -> None: + """Replace the current card with the plans list at ``offset``.""" + text, markup = await self._plans_view_for(offset) + await self._safe_edit(query, text, markup) + + async def _edit_plan_detail(self, query: Any, plan_id: str) -> None: + """Render a plan's detail (HTML) into the card in place.""" + plan = await self.router.get_plan(plan_id) + if not plan: + await self._edit_plans_list(query) + return + text, markup = build_plan_detail_view(plan, tzname=self.config.timezone) + await self._safe_edit(query, text, markup, parse_mode=ParseMode.HTML) + + async def _handle_plan_button(self, query: Any) -> None: + """Handle /plans inline-keyboard presses. + + callback_data forms: + ``plan:list:`` — (re)show the review queue at a page + ``plan:view:`` — show a plan's detail + action buttons + ``plan:approve:`` — confirm-approve card + ``plan:approveok:`` — approve (spawns an implementation session) + ``plan:decline:`` — confirm-decline card + ``plan:declineok:`` — decline (closes the task as done) + ``plan:revise:`` — prompt (ForceReply) for revision feedback + """ + if not query.message: + await query.answer() + return + parts = query.data.split(":", 2) + action = parts[1] if len(parts) > 1 else "" + arg = parts[2] if len(parts) > 2 else "" + + if action == "list": + try: + off = max(0, int(arg)) + except ValueError: + off = 0 + await query.answer() + await self._edit_plans_list(query, off) + return + + if action == "view": + await query.answer() + await self._edit_plan_detail(query, arg) + return + + if action in ("approve", "decline"): + plan = await self.router.get_plan(arg) + if not plan: + await query.answer("That plan is no longer available", show_alert=True) + await self._edit_plans_list(query) + return + if plan.get("status") != "pending": + await query.answer( + f"Plan is {plan.get('status')} — no longer actionable", + show_alert=True, + ) + await self._edit_plan_detail(query, arg) + return + await query.answer() + text, markup = build_plan_confirm_view(plan, action) + await self._safe_edit(query, text, markup) + return + + if action == "approveok": + await self._do_plan_approve(query, arg) + return + if action == "declineok": + await self._do_plan_decline(query, arg) + return + if action == "revise": + await self._start_plan_revise(query, arg) + return + + await query.answer() + + async def _do_plan_approve(self, query: Any, plan_id: str) -> None: + """Approve via the shared service, then confirm on the card.""" + from nerve.agent.plan_service import ( + PlanNotFound, + PlanNotPending, + TaskNotFound, + approve_plan, + ) + try: + result = await approve_plan( + db=self.router.engine.db, engine=self.router.engine, plan_id=plan_id, + ) + except (PlanNotFound, TaskNotFound): + await query.answer("That plan is no longer available", show_alert=True) + await self._edit_plans_list(query) + return + except PlanNotPending as exc: + await query.answer(str(exc), show_alert=True) + await self._edit_plan_detail(query, plan_id) + return + await query.answer("Approved — implementation started") + impl = _html.escape(str(result.get("impl_session_id") or "?")) + text = ( + "✅ Plan approved.\n" + f"Implementation session {impl} started." + ) + markup = InlineKeyboardMarkup( + [[InlineKeyboardButton("🗒 Plans", callback_data="plan:list:0")]] + ) + await self._safe_edit(query, text, markup, parse_mode=ParseMode.HTML) + + async def _do_plan_decline(self, query: Any, plan_id: str) -> None: + """Decline via the shared service, then confirm on the card.""" + from nerve.agent.plan_service import ( + PlanNotFound, + PlanNotPending, + TaskNotFound, + decline_plan, + ) + try: + await decline_plan( + db=self.router.engine.db, engine=self.router.engine, plan_id=plan_id, + ) + except (PlanNotFound, TaskNotFound): + await query.answer("That plan is no longer available", show_alert=True) + await self._edit_plans_list(query) + return + except PlanNotPending as exc: + await query.answer(str(exc), show_alert=True) + await self._edit_plan_detail(query, plan_id) + return + await query.answer("Declined — task closed") + text = "🗑 Plan declined.\nThe related task was closed as done." + markup = InlineKeyboardMarkup( + [[InlineKeyboardButton("🗒 Plans", callback_data="plan:list:0")]] + ) + await self._safe_edit(query, text, markup, parse_mode=ParseMode.HTML) + + async def _start_plan_revise(self, query: Any, plan_id: str) -> None: + """Ask for revision feedback via a ForceReply prompt. + + The next message that *replies to* this prompt is captured as feedback + in ``_handle_message`` and dispatched to the planner — Telegram's + native way to collect a text argument after a button press. + """ + plan = await self.router.get_plan(plan_id) + if not plan or plan.get("status") != "pending": + await query.answer( + "Only pending plans can be revised", show_alert=True, + ) + await self._edit_plan_detail(query, plan_id) + return + await query.answer() + title = plan.get("task_title") or plan.get("task_id") or plan_id + prompt = await query.message.reply_text( + f"✍️ Reply to this message with revision feedback for the plan " + f"“{title}”.\nThe planner will produce a new version from your notes.", + reply_markup=ForceReply(input_field_placeholder="What should change?"), + ) + self._pending_plan_revision[query.message.chat.id] = ( + plan_id, prompt.message_id, + ) + + async def _submit_plan_revision( + self, update: Update, plan_id: str, feedback: str, + ) -> None: + """Dispatch captured revision feedback to the planner session.""" + from nerve.agent.plan_service import ( + PlanNotFound, + PlanNotPending, + TaskNotFound, + request_plan_revision, + ) + try: + result = await request_plan_revision( + db=self.router.engine.db, engine=self.router.engine, + plan_id=plan_id, feedback=feedback, + ) + except (PlanNotFound, TaskNotFound): + await update.message.reply_text("That plan no longer exists.") + return + except (PlanNotPending, ValueError) as exc: + await update.message.reply_text(str(exc)) + return + await update.message.reply_text( + f"✍️ Revision requested for plan {plan_id}. Feedback sent to the " + f"planner ({result['session_id']}); a new version will appear in /plans." + ) + + async def _maybe_consume_plan_revision(self, update: Update) -> bool: + """Consume a reply to a /plans revise prompt as revision feedback. + + Returns True (message handled — do not route to the agent) only when + this chat has a pending revise prompt AND this message replies to that + exact prompt. Any other message falls through to normal routing. + """ + chat = update.effective_chat + msg = update.message + if not chat or not msg: + return False + pending = self._pending_plan_revision.get(chat.id) + if not pending: + return False + reply_to = getattr(msg, "reply_to_message", None) + if not reply_to or reply_to.message_id != pending[1]: + return False + plan_id, _prompt_id = pending + self._pending_plan_revision.pop(chat.id, None) + feedback = (msg.text or "").strip() + if not feedback: + await msg.reply_text("No feedback given — revision cancelled.") + return True + await self._submit_plan_revision(update, plan_id, feedback) + return True + async def _handle_star(self, update: Update, context: Any) -> None: """Handle /star — star the current session so it never auto-closes.""" await self._set_current_starred(update, True) @@ -1557,6 +1984,11 @@ async def _handle_message(self, update: Update, context: Any) -> None: if not self._is_authorized(update.effective_user.id): return + # A reply to a /plans "Revise" ForceReply prompt is revision feedback, + # not a message for the agent. Intercept before any routing. + if await self._maybe_consume_plan_revision(update): + return + # Media group (album) — collect all parts before processing if update.message.media_group_id: await self._collect_media_group(update) @@ -1940,6 +2372,11 @@ async def _handle_callback_query(self, update: Update, context: Any) -> None: await self._handle_session_button(query) return + # /plans inline keyboard: list/view/approve/decline/revise. + if query.data.startswith("plan:"): + await self._handle_plan_button(query) + return + # Parse callback_data: "notif:{notification_id}:{answer}" parts = query.data.split(":", 2) if len(parts) < 3 or parts[0] != "notif": diff --git a/nerve/gateway/routes/plans.py b/nerve/gateway/routes/plans.py index e3b2f655..41e74a42 100644 --- a/nerve/gateway/routes/plans.py +++ b/nerve/gateway/routes/plans.py @@ -2,9 +2,7 @@ from __future__ import annotations -import asyncio import logging -import uuid from datetime import datetime, timezone from fastapi import APIRouter, Depends, HTTPException @@ -14,15 +12,12 @@ PlanNotFound, PlanNotPending, TaskNotFound, + approve_plan, + decline_plan, request_plan_revision, ) -from nerve.config import get_config from nerve.gateway.auth import require_auth -from nerve.gateway.routes._deps import ( - build_route_tool_context, - get_deps, - get_tool_registry, -) +from nerve.gateway.routes._deps import get_deps logger = logging.getLogger(__name__) @@ -60,6 +55,25 @@ async def get_plan(plan_id: str, user: dict = Depends(require_auth)): @router.patch("/api/plans/{plan_id}") async def update_plan(plan_id: str, req: PlanUpdateRequest, user: dict = Depends(require_auth)): deps = get_deps() + + # Decline is the only status transition the UI drives through PATCH. + # Delegate to the shared service so the WebUI, MCP tool, and Telegram + # all mark the plan declined and close the task identically. + if req.status == "declined": + try: + await decline_plan( + db=deps.db, engine=deps.engine, + plan_id=plan_id, feedback=req.feedback, + ) + except PlanNotFound: + raise HTTPException(status_code=404, detail="Plan not found") + except TaskNotFound: + raise HTTPException(status_code=404, detail="Task not found") + except PlanNotPending as exc: + raise HTTPException(status_code=409, detail=str(exc)) + return {"plan_id": plan_id, "updated": True} + + # Generic field update (e.g. feedback-only, or a non-decline status). plan = await deps.db.get_plan(plan_id) if not plan: raise HTTPException(status_code=404, detail="Plan not found") @@ -74,20 +88,6 @@ async def update_plan(plan_id: str, req: PlanUpdateRequest, user: dict = Depends if fields: await deps.db.update_plan(plan_id, **fields) - # On decline: mark the related task as done with a note explaining - # the closure. The user-supplied feedback is optional — if absent, - # leave a generic comment. - if req.status == "declined": - if req.feedback: - note = f"Plan {plan_id} declined — {req.feedback}" - else: - note = f"Related plan {plan_id} was closed without a specified reason" - await get_tool_registry().invoke( - "task_done", - build_route_tool_context(), - {"task_id": plan["task_id"], "note": note}, - ) - return {"plan_id": plan_id, "updated": True} @@ -121,122 +121,31 @@ async def revise_plan(plan_id: str, req: PlanReviseRequest, user: dict = Depends @router.post("/api/plans/{plan_id}/approve") -async def approve_plan( +async def approve_plan_route( plan_id: str, user: dict = Depends(require_auth), ): - """Approve a plan and spawn an implementation session.""" - deps = get_deps() - plan = await deps.db.get_plan(plan_id) - if not plan: - raise HTTPException(status_code=404, detail="Plan not found") + """Approve a plan and spawn an implementation session. - # Guard: only pending plans can be approved - if plan["status"] != "pending": - raise HTTPException( - status_code=409, - detail=f"Plan is '{plan['status']}', only 'pending' plans can be approved", + Thin wrapper around the shared ``approve_plan`` service so the WebUI, + MCP tool, and Telegram all spawn identically-briefed implementation + sessions. Helper exceptions map to HTTP status codes. + """ + deps = get_deps() + try: + result = await approve_plan( + db=deps.db, engine=deps.engine, plan_id=plan_id, ) - - task = await deps.db.get_task(plan["task_id"]) - if not task: + except PlanNotFound: + raise HTTPException(status_code=404, detail="Plan not found") + except TaskNotFound: raise HTTPException(status_code=404, detail="Task not found") - - now = datetime.now(timezone.utc).isoformat() - plan_type = plan.get("plan_type", "generic") - - # Mark plan as implementing immediately (prevents double-approve) - await deps.db.update_plan(plan_id, status="implementing", reviewed_at=now) - - # Create implementation session (visible in Chat UI) - impl_session_id = f"impl-{str(uuid.uuid4())[:8]}" - await deps.engine.sessions.get_or_create( - impl_session_id, title=f"Implement: {task['title']}", source="web", - ) - await deps.db.update_plan(plan_id, impl_session_id=impl_session_id) - - # Update task status + note - await get_tool_registry().invoke( - "task_update", - build_route_tool_context(), - { - "task_id": plan["task_id"], - "status": "in_progress", - "note": f"Plan approved — implementation started (session: {impl_session_id})", - }, - ) - - # Read task file content for the implementation prompt - config = get_config() - task_content = "" - if task.get("file_path"): - task_file = config.workspace / task["file_path"] - if task_file.exists(): - task_content = await asyncio.to_thread( - task_file.read_text, encoding="utf-8", - ) - - # Build implementation prompt — skill-aware - if plan_type in ("skill-create", "skill-update"): - prompt = ( - f"You are implementing an approved plan for a skill task.\n\n" - f"## Task: {task['title']}\n\n" - f"### Task Content\n{task_content}\n\n" - f"## Approved Plan\n{plan['content']}\n\n" - f"## Instructions\n" - ) - if plan_type == "skill-create": - prompt += ( - "The plan contains a skill specification. " - "Use the `skill_create` tool to create the skill. " - "Extract the name, description, and content from the plan. " - "If the plan contains a full SKILL.md with frontmatter, parse out the name and description " - "from the frontmatter and use the body as the content.\n" - ) - else: - prompt += ( - "The plan contains a skill revision. " - "Use the `skill_update` tool to update the existing skill. " - "Pass the skill ID (directory name) as the name parameter and the full SKILL.md content " - "(frontmatter + body).\n" - ) - prompt += ( - "\nAfter the skill is created/updated, mark the task as done using " - "`task_done` with a note describing what was done.\n" - ) - else: - prompt = ( - f"You are implementing an approved plan for a task.\n\n" - f"## Task: {task['title']}\n\n" - f"### Task Content\n{task_content}\n\n" - f"## Approved Plan\n{plan['content']}\n\n" - f"## Instructions\n" - f"Follow the plan step by step. You have full tool access.\n" - f"After implementation, verify your changes work correctly.\n" - f"If you encounter issues not covered by the plan, use your judgment or ask the user.\n" - ) - - # Spawn implementation in background with error handling. Register - # the task with the engine so a manual /stop can cancel a stuck impl - # session (without registration, the asyncio.Task is invisible to - # `engine.stop_session` and the only way to recover is a daemon - # restart). - async def _run_impl(): - try: - await deps.engine.run( - session_id=impl_session_id, user_message=prompt, source="web", - ) - except Exception: - logger.exception("Implementation session %s failed", impl_session_id) - try: - await deps.db.update_plan(plan_id, status="failed") - except Exception: - logger.exception("Failed to mark plan %s as failed", plan_id) - - impl_task = asyncio.create_task(_run_impl()) - deps.engine.register_task(impl_session_id, impl_task) - - return {"plan_id": plan_id, "impl_session_id": impl_session_id} + except PlanNotPending as exc: + raise HTTPException(status_code=409, detail=str(exc)) + return { + "plan_id": result["plan_id"], + "impl_session_id": result["impl_session_id"], + } @router.get("/api/tasks/{task_id}/plans") diff --git a/tests/test_plan_actions.py b/tests/test_plan_actions.py new file mode 100644 index 00000000..fd85a77f --- /dev/null +++ b/tests/test_plan_actions.py @@ -0,0 +1,171 @@ +"""Tests for the shared plan approve/decline helpers (``nerve.agent.plan_service``). + +``approve_plan`` and ``decline_plan`` back three surfaces — the HTTP routes +(WebUI), the MCP ``plan_approve``/``plan_decline`` tools, and the Telegram +``/plans`` command. These pin the single behaviour contract so the surfaces +can't drift apart (the same rationale as ``test_plan_revise.py``). +""" + +from __future__ import annotations + +import asyncio + +import pytest + +from nerve.agent import tools as tools_mod +from nerve.agent.plan_service import ( + PlanNotFound, + PlanNotPending, + TaskNotFound, + approve_plan, + decline_plan, +) +from nerve.db import Database + + +class FakeSessionManager: + """Records get_or_create calls so tests can assert the impl session.""" + + def __init__(self) -> None: + self.calls: list[dict] = [] + + async def get_or_create( + self, session_id, title=None, source="web", metadata=None, + ) -> dict: + self.calls.append({"session_id": session_id, "title": title, "source": source}) + return {"id": session_id, "title": title or session_id, "source": source} + + +class FakeEngine: + """Mimics AgentEngine.run + .sessions + .register_task for approve tests.""" + + def __init__(self) -> None: + self.sessions = FakeSessionManager() + self.runs: list[dict] = [] + self.registered: list[str] = [] + self.run_event = asyncio.Event() + + async def run(self, session_id, user_message, source="web") -> None: + self.runs.append( + {"session_id": session_id, "user_message": user_message, "source": source} + ) + self.run_event.set() + + def register_task(self, session_id, task) -> None: + self.registered.append(session_id) + + +async def _setup( + db: Database, tmp_path, *, plan_type: str = "generic", status: str = "pending", +) -> tuple[FakeEngine, str]: + task_id = "t-act" + file_path = "task.md" + (tmp_path / file_path).write_text("# Demo task\n\nBody text.\n", encoding="utf-8") + await db.upsert_task( + task_id=task_id, file_path=file_path, title="Demo task", + status="pending", content=(tmp_path / file_path).read_text(), + ) + await db.create_plan( + plan_id="plan-act", task_id=task_id, content="step one; step two", + session_id="sess-proposer", version=1, plan_type=plan_type, + ) + if status != "pending": + await db.update_plan("plan-act", status=status) + + engine = FakeEngine() + tools_mod.init_tools(workspace=tmp_path, db=db, engine=engine) + return engine, task_id + + +@pytest.mark.asyncio +class TestApprovePlan: + async def test_spawns_impl_marks_implementing_and_moves_task(self, db, tmp_path): + engine, task_id = await _setup(db, tmp_path) + + result = await approve_plan(db=db, engine=engine, plan_id="plan-act") + await asyncio.wait_for(engine.run_event.wait(), timeout=1.0) + + impl = result["impl_session_id"] + assert impl.startswith("impl-") + assert result["plan_id"] == "plan-act" + assert result["task_id"] == task_id + + plan = await db.get_plan("plan-act") + assert plan["status"] == "implementing" + assert plan["impl_session_id"] == impl + + task = await db.get_task(task_id) + assert task["status"] == "in_progress" + + # Impl session created, registered with the engine, run dispatched. + assert engine.sessions.calls[0]["session_id"] == impl + assert engine.registered == [impl] + assert len(engine.runs) == 1 + prompt = engine.runs[0]["user_message"] + assert "step one; step two" in prompt # plan content + assert "Demo task" in prompt # task title + assert "Body text." in prompt # task file content threaded in + + async def test_skill_create_gets_skill_prompt(self, db, tmp_path): + engine, _ = await _setup(db, tmp_path, plan_type="skill-create") + await approve_plan(db=db, engine=engine, plan_id="plan-act") + await asyncio.wait_for(engine.run_event.wait(), timeout=1.0) + assert "skill_create" in engine.runs[0]["user_message"] + + async def test_refuses_non_pending(self, db, tmp_path): + engine, _ = await _setup(db, tmp_path, status="implementing") + with pytest.raises(PlanNotPending): + await approve_plan(db=db, engine=engine, plan_id="plan-act") + assert engine.runs == [] + + async def test_raises_plan_not_found(self, db, tmp_path): + engine, _ = await _setup(db, tmp_path) + with pytest.raises(PlanNotFound): + await approve_plan(db=db, engine=engine, plan_id="plan-missing") + assert engine.runs == [] + + async def test_raises_task_not_found(self, db, tmp_path): + engine, _ = await _setup(db, tmp_path) + await db.db.execute("DELETE FROM tasks WHERE id = ?", ("t-act",)) + await db.db.commit() + with pytest.raises(TaskNotFound): + await approve_plan(db=db, engine=engine, plan_id="plan-act") + assert engine.runs == [] + + +@pytest.mark.asyncio +class TestDeclinePlan: + async def test_marks_declined_and_closes_task(self, db, tmp_path): + engine, task_id = await _setup(db, tmp_path) + + result = await decline_plan( + db=db, engine=engine, plan_id="plan-act", feedback="not now", + ) + assert result["status"] == "declined" + assert result["feedback"] == "not now" + + plan = await db.get_plan("plan-act") + assert plan["status"] == "declined" + assert plan["feedback"] == "not now" + + task = await db.get_task(task_id) + assert task["status"] == "done" + + async def test_without_feedback_still_closes(self, db, tmp_path): + engine, task_id = await _setup(db, tmp_path) + result = await decline_plan(db=db, engine=engine, plan_id="plan-act") + assert result["feedback"] == "" + plan = await db.get_plan("plan-act") + assert plan["status"] == "declined" + task = await db.get_task(task_id) + assert task["status"] == "done" + + async def test_refuses_non_pending(self, db, tmp_path): + engine, _ = await _setup(db, tmp_path, status="declined") + with pytest.raises(PlanNotPending): + await decline_plan(db=db, engine=engine, plan_id="plan-act") + + async def test_raises_plan_not_found(self, db, tmp_path): + engine, _ = await _setup(db, tmp_path) + with pytest.raises(PlanNotFound): + await decline_plan(db=db, engine=engine, plan_id="plan-missing") diff --git a/tests/test_telegram_plans.py b/tests/test_telegram_plans.py new file mode 100644 index 00000000..33eccc30 --- /dev/null +++ b/tests/test_telegram_plans.py @@ -0,0 +1,242 @@ +"""Tests for the /plans inline-keyboard builders (nerve.channels.telegram).""" + +import pytest + +from nerve.channels.telegram import ( + _PLANS_PAGE_SIZE, + build_plan_confirm_view, + build_plan_detail_view, + build_plans_view, +) + + +def _flat(markup): + return [b for row in markup.inline_keyboard for b in row] + + +def _cbs(markup): + return [b.callback_data for b in _flat(markup)] + + +# --- list view (build_plans_view) ----------------------------------------- # + +def test_plan_ids_ride_in_callback_data_and_refresh_always_present(): + plans = [ + {"id": "plan-aaaa1111", "task_title": "Add dark mode", "status": "pending", "version": 1}, + {"id": "plan-bbbb2222", "task_title": "Cache API", "status": "implementing", "version": 1}, + ] + text, markup = build_plans_view(plans) + by_cb = {b.callback_data: b for b in _flat(markup)} + # One tap-to-open button per plan (id carried, no copy-paste). + assert by_cb["plan:view:plan-aaaa1111"].text == "🟠 Add dark mode" + assert by_cb["plan:view:plan-bbbb2222"].text == "⚙️ Cache API" + # Refresh is always the last button so the keyboard is never empty. + assert markup.inline_keyboard[-1][0].callback_data == "plan:list:0" + assert "pending" in text + + +def test_version_suffix_only_when_above_one(): + plans = [{"id": "plan-v", "task_title": "T", "status": "pending", "version": 3}] + _text, markup = build_plans_view(plans) + assert _flat(markup)[0].text == "🟠 T v3" + + +def test_empty_queue_has_only_refresh(): + text, markup = build_plans_view([]) + assert _cbs(markup) == ["plan:list:0"] + assert "No plans awaiting review" in text + + +def test_long_title_truncated(): + plans = [{"id": "plan-x", "task_title": "x" * 100, "status": "pending", "version": 1}] + _text, markup = build_plans_view(plans) + label = _flat(markup)[0].text + assert label.endswith("…") + assert len(label) <= 42 # emoji + space + clipped title + + +def test_oversized_callback_data_is_skipped(): + huge = "plan-" + "z" * 70 # plan:view: > 64 bytes → must be dropped + plans = [ + {"id": huge, "task_title": "too big", "status": "pending", "version": 1}, + {"id": "plan-ok", "task_title": "ok", "status": "pending", "version": 1}, + ] + _text, markup = build_plans_view(plans) + cbs = _cbs(markup) + assert f"plan:view:{huge}" not in cbs + assert "plan:view:plan-ok" in cbs + + +def test_page_size_caps_rows_but_keeps_refresh(): + plans = [ + {"id": f"plan-{n:06d}", "task_title": f"P{n}", "status": "pending", "version": 1} + for n in range(50) + ] + _text, markup = build_plans_view(plans, has_next=True) + view_btns = [c for c in _cbs(markup) if c.startswith("plan:view:")] + assert len(view_btns) == _PLANS_PAGE_SIZE + assert "plan:list:0" in _cbs(markup) # refresh still present + + +def test_first_page_offers_more_not_prev(): + plans = [{"id": f"plan-{n:03d}", "task_title": f"P{n}", "status": "pending", "version": 1} + for n in range(_PLANS_PAGE_SIZE)] + _text, markup = build_plans_view(plans, offset=0, has_prev=False, has_next=True) + cbs = _cbs(markup) + assert f"plan:list:{_PLANS_PAGE_SIZE}" in cbs # ➡️ More → page 2 + # No ⬅️ Prev target other than the always-present refresh (plan:list:0). + assert "⬅️ Prev" not in [b.text for b in _flat(markup)] + + +def test_middle_page_offers_prev_and_more(): + off = _PLANS_PAGE_SIZE + plans = [{"id": f"plan-{n:03d}", "task_title": f"P{n}", "status": "pending", "version": 1} + for n in range(_PLANS_PAGE_SIZE)] + text, markup = build_plans_view(plans, offset=off, has_prev=True, has_next=True) + cbs = _cbs(markup) + assert f"plan:list:{max(0, off - _PLANS_PAGE_SIZE)}" in cbs # ⬅️ Prev + assert f"plan:list:{off + _PLANS_PAGE_SIZE}" in cbs # ➡️ More + assert "Page 2" in text + + +# --- detail view (build_plan_detail_view) --------------------------------- # + +_PENDING = { + "id": "plan-detail1", "task_title": "Add dark mode", "task_id": "t1", + "status": "pending", "version": 2, "plan_type": "generic", + "content": "1. add toggle\n2. persist choice", + "created_at": "2026-08-20T09:00:00+00:00", +} + + +def test_detail_pending_shows_all_actions(): + text, markup = build_plan_detail_view(_PENDING, tzname="UTC") + cbs = _cbs(markup) + assert cbs == [ + "plan:approve:plan-detail1", + "plan:decline:plan-detail1", + "plan:revise:plan-detail1", + "plan:list:0", + ] + assert "
" in text # body collapses/expands natively + assert "Add dark mode" in text + assert "v2" in text + + +def test_detail_non_pending_is_read_only_and_shows_impl(): + plan = { + "id": "plan-impl", "task_title": "Cache API", "status": "implementing", + "version": 1, "content": "do it", "impl_session_id": "impl-1234abcd", + "created_at": "2026-08-20T09:00:00+00:00", + } + text, markup = build_plan_detail_view(plan, tzname="UTC") + # No approve/decline/revise on a non-pending plan — only back-to-list. + assert _cbs(markup) == ["plan:list:0"] + assert "impl-1234abcd" in text + + +def test_detail_escapes_html_in_content(): + plan = {**_PENDING, "content": "&danger"} + text, _m = build_plan_detail_view(plan, tzname="UTC") + assert "<b>&danger</b>" in text + + +def test_detail_shows_revision_feedback(): + plan = {**_PENDING, "feedback": "please add tests"} + text, _m = build_plan_detail_view(plan, tzname="UTC") + assert "please add tests" in text + + +def test_detail_stays_within_telegram_limit(): + plan = {**_PENDING, "content": "y" * 8000} + text, _m = build_plan_detail_view(plan, tzname="UTC") + assert len(text) <= 4096 + + +# --- confirm view (build_plan_confirm_view) ------------------------------- # + +def test_confirm_approve_buttons(): + text, markup = build_plan_confirm_view(_PENDING, "approve") + cbs = _cbs(markup) + assert "plan:approveok:plan-detail1" in cbs + assert "plan:view:plan-detail1" in cbs # ◀️ Back to detail + assert "implementation session" in text.lower() + + +def test_confirm_decline_buttons(): + text, markup = build_plan_confirm_view(_PENDING, "decline") + cbs = _cbs(markup) + assert "plan:declineok:plan-detail1" in cbs + assert "plan:view:plan-detail1" in cbs + assert "closes the task" in text.lower() + + +# --- _plans_view_for paging (pending-first, implementing after) ----------- # + +class _FakeRouter: + """Stands in for the channel router: returns plans by status like the + store would (each already newest-first).""" + + def __init__(self, pending, implementing=None): + self._pending = pending + self._impl = implementing or [] + self.calls = [] + + async def list_plans(self, status=None, limit=100): + self.calls.append((status, limit)) + if status == "pending": + return list(self._pending) + if status == "implementing": + return list(self._impl) + return list(self._pending) + list(self._impl) + + +def _make_channel(router): + from nerve.channels.telegram import TelegramChannel + ch = TelegramChannel.__new__(TelegramChannel) # bypass __init__; only .router needed + ch.router = router + return ch + + +@pytest.mark.asyncio +async def test_plans_view_lists_pending_then_implementing(): + router = _FakeRouter( + pending=[{"id": "plan-p1", "task_title": "P1", "status": "pending", "version": 1}], + implementing=[{"id": "plan-i1", "task_title": "I1", "status": "implementing", "version": 1}], + ) + ch = _make_channel(router) + _text, markup = await ch._plans_view_for() + cbs = _cbs(markup) + # pending first, implementing next, refresh last. + assert cbs == ["plan:view:plan-p1", "plan:view:plan-i1", "plan:list:0"] + + +@pytest.mark.asyncio +async def test_plans_view_paginates(): + pending = [ + {"id": f"plan-{n:03d}", "task_title": f"P{n}", "status": "pending", "version": 1} + for n in range(_PLANS_PAGE_SIZE * 2 + 1) + ] + ch = _make_channel(_FakeRouter(pending=pending)) + + # Page 1: More but no Prev. + _t, m1 = await ch._plans_view_for(0) + cbs1 = _cbs(m1) + assert len([c for c in cbs1 if c.startswith("plan:view:")]) == _PLANS_PAGE_SIZE + assert f"plan:list:{_PLANS_PAGE_SIZE}" in cbs1 + + # Last page: Prev but no More. + _t, m3 = await ch._plans_view_for(_PLANS_PAGE_SIZE * 2) + cbs3 = _cbs(m3) + assert len([c for c in cbs3 if c.startswith("plan:view:")]) == 1 # remainder + assert f"plan:list:{_PLANS_PAGE_SIZE}" in cbs3 # ⬅️ Prev → page 2 + # No ➡️ More button target beyond the last page. + assert not any(c == f"plan:list:{_PLANS_PAGE_SIZE * 3}" for c in cbs3) + + +@pytest.mark.asyncio +async def test_plans_view_empty(): + ch = _make_channel(_FakeRouter(pending=[])) + text, markup = await ch._plans_view_for() + assert _cbs(markup) == ["plan:list:0"] + assert "No plans awaiting review" in text