LZH-YS1998
288 lines
11 KiB
Python
288 lines
11 KiB
Python
"""Interactive task actions for the CLI board."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import uuid
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from opc.core.models import Task, TaskStatus
|
|
from opc.layer2_organization.work_item_transition import apply_task_status_transition
|
|
from opc.presentation.kanban import column_to_task_status
|
|
from opc.plugins.office_ui.services.factory import OfficeServiceFactory
|
|
|
|
if TYPE_CHECKING:
|
|
from .engine_facade import EngineFacade
|
|
|
|
|
|
class BoardActions:
|
|
"""Mutating operations exposed to the TUI."""
|
|
|
|
def __init__(self, facade: "EngineFacade", project_id: str | None = None) -> None:
|
|
self.facade = facade
|
|
self.project_id = project_id
|
|
self._task_locks: dict[str, asyncio.Lock] = {}
|
|
self._task_bg_map: dict[str, asyncio.Task[Any]] = {}
|
|
|
|
@property
|
|
def _project_id(self) -> str:
|
|
return self.project_id or self.facade.project_id or "default"
|
|
|
|
async def _run_office_service(self, operation):
|
|
engine = await self.facade.ensure_ready()
|
|
async with OfficeServiceFactory(
|
|
config=getattr(engine, "config", None),
|
|
project_id=self._project_id,
|
|
on_progress=getattr(self.facade, "_progress_callback", None),
|
|
on_runtime_event=getattr(self.facade, "_event_callback", None),
|
|
) as services:
|
|
return await operation(services)
|
|
|
|
async def create_task(
|
|
self,
|
|
*,
|
|
title: str,
|
|
description: str = "",
|
|
auto_run: bool = False,
|
|
initial_message: str | None = None,
|
|
mode: str = "task",
|
|
company_profile: str | None = None,
|
|
) -> Task:
|
|
engine = await self.facade.ensure_ready()
|
|
if not engine.store:
|
|
raise RuntimeError("OPC store is not available.")
|
|
|
|
normalized_title = str(title or "").strip() or "New Chat"
|
|
normalized_description = str(description or "").strip()
|
|
result = await self._run_office_service(
|
|
lambda svc: svc.session.create(
|
|
project_id=self._project_id,
|
|
title=normalized_title,
|
|
description=normalized_description,
|
|
exec_mode=mode,
|
|
company_profile=company_profile,
|
|
interface="cli_board",
|
|
)
|
|
)
|
|
task_id = str(result.payload.get("task_id", "") or "")
|
|
task = await self._get_task(task_id)
|
|
|
|
if auto_run:
|
|
prompt = (initial_message or f"{normalized_title}\n{normalized_description}").strip()
|
|
if prompt:
|
|
await self.send_session_message(
|
|
task.id,
|
|
prompt,
|
|
mode=mode,
|
|
company_profile=company_profile,
|
|
allow_terminal=True,
|
|
)
|
|
return task
|
|
|
|
async def run_task(
|
|
self,
|
|
task_id: str,
|
|
*,
|
|
mode: str = "task",
|
|
company_profile: str | None = None,
|
|
) -> str:
|
|
task = await self._get_task(task_id)
|
|
prompt = f"{task.title}\n{task.description}".strip()
|
|
if not prompt:
|
|
raise ValueError("Selected task has no title or description to run.")
|
|
return await self.send_session_message(
|
|
task_id,
|
|
prompt,
|
|
mode=mode,
|
|
company_profile=company_profile,
|
|
allow_terminal=True,
|
|
)
|
|
|
|
async def retry_task(
|
|
self,
|
|
task_id: str,
|
|
*,
|
|
mode: str = "task",
|
|
company_profile: str | None = None,
|
|
) -> str:
|
|
return await self.run_task(task_id, mode=mode, company_profile=company_profile)
|
|
|
|
async def send_session_message(
|
|
self,
|
|
task_id: str,
|
|
content: str,
|
|
*,
|
|
mode: str = "task",
|
|
company_profile: str | None = None,
|
|
allow_terminal: bool = False,
|
|
) -> str:
|
|
engine = await self.facade.ensure_ready()
|
|
if not engine.store:
|
|
raise RuntimeError("OPC store is not available.")
|
|
|
|
async with self._get_task_lock(task_id):
|
|
if existing := self._task_bg_map.get(task_id):
|
|
if not existing.done() and existing is not asyncio.current_task():
|
|
raise RuntimeError("This task is already running.")
|
|
|
|
task = await self._get_task(task_id)
|
|
if not allow_terminal and task.status in {TaskStatus.CANCELLED, TaskStatus.DONE}:
|
|
raise ValueError("Cannot send messages to a completed or cancelled task.")
|
|
|
|
if not task.session_id:
|
|
task.session_id = str(uuid.uuid4())
|
|
if engine.memory:
|
|
await engine.memory.ensure_session(
|
|
task.session_id,
|
|
project_id=self._project_id,
|
|
title=task.title,
|
|
mode="primary",
|
|
metadata={"source": "cli_board"},
|
|
)
|
|
|
|
current_task = asyncio.current_task()
|
|
if current_task is not None:
|
|
self._task_bg_map[task_id] = current_task
|
|
|
|
try:
|
|
normalized_mode = str(mode or "task").strip().lower()
|
|
send_mode = "company" if normalized_mode in {"company", "custom", "org"} else "task"
|
|
send_profile = "custom" if normalized_mode in {"custom", "org"} else company_profile
|
|
result = await self._run_office_service(
|
|
lambda svc: svc.session.send(
|
|
project_id=self._project_id,
|
|
task_id=task_id,
|
|
content=str(content or "").strip(),
|
|
mode=send_mode,
|
|
company_profile=send_profile,
|
|
)
|
|
)
|
|
return str(result.payload.get("response", "") or "")
|
|
except asyncio.CancelledError:
|
|
await self._mark_related_tasks(task_id, TaskStatus.CANCELLED)
|
|
await self._resolve_related_checkpoints(task_id, status="cancelled")
|
|
raise
|
|
finally:
|
|
if self._task_bg_map.get(task_id) is current_task:
|
|
self._task_bg_map.pop(task_id, None)
|
|
|
|
async def move_task(self, task_id: str, column_id: str) -> Task:
|
|
engine = await self.facade.ensure_ready()
|
|
if not engine.store:
|
|
raise RuntimeError("OPC store is not available.")
|
|
|
|
task = await self._get_task(task_id)
|
|
target_status = column_to_task_status(column_id)
|
|
if target_status is None:
|
|
raise ValueError(f"Unsupported target column: {column_id}")
|
|
if task.status in {TaskStatus.DONE, TaskStatus.FAILED, TaskStatus.CANCELLED} and target_status not in {
|
|
TaskStatus.DONE,
|
|
TaskStatus.FAILED,
|
|
TaskStatus.CANCELLED,
|
|
}:
|
|
raise ValueError(f"Cannot move terminal task {task.status.value} back to {column_id}.")
|
|
await self._run_office_service(lambda svc: svc.kanban.move_task(project_id=self._project_id, task_id=task_id, column_id=column_id))
|
|
return await self._get_task(task_id)
|
|
|
|
async def complete_task(self, task_id: str) -> Task:
|
|
engine = await self.facade.ensure_ready()
|
|
if not engine.store:
|
|
raise RuntimeError("OPC store is not available.")
|
|
await self._run_office_service(lambda svc: svc.session.complete(project_id=self._project_id, task_id=task_id))
|
|
return await self._get_task(task_id)
|
|
|
|
async def cancel_task(self, task_id: str) -> None:
|
|
engine = await self.facade.ensure_ready()
|
|
if not engine.store:
|
|
raise RuntimeError("OPC store is not available.")
|
|
|
|
background = self._task_bg_map.get(task_id)
|
|
if background and not background.done() and background is not asyncio.current_task():
|
|
background.cancel()
|
|
await asyncio.sleep(0)
|
|
|
|
await self._run_office_service(lambda svc: svc.session.stop(project_id=self._project_id, task_id=task_id))
|
|
await self._resolve_related_checkpoints(task_id, status="cancelled")
|
|
|
|
async def approve_checkpoint(self, task_id: str, *, approved: bool = True, reply: str | None = None) -> str:
|
|
if reply is not None:
|
|
message = reply
|
|
else:
|
|
message = "approve" if approved else "deny"
|
|
return await self.send_session_message(
|
|
task_id,
|
|
message,
|
|
allow_terminal=True,
|
|
)
|
|
|
|
def is_running(self, task_id: str) -> bool:
|
|
task = self._task_bg_map.get(task_id)
|
|
return bool(task and not task.done())
|
|
|
|
def _get_task_lock(self, task_id: str) -> asyncio.Lock:
|
|
lock = self._task_locks.get(task_id)
|
|
if lock is None:
|
|
lock = asyncio.Lock()
|
|
self._task_locks[task_id] = lock
|
|
return lock
|
|
|
|
async def _get_task(self, task_id: str) -> Task:
|
|
engine = await self.facade.ensure_ready()
|
|
if not engine.store:
|
|
raise RuntimeError("OPC store is not available.")
|
|
task = await engine.store.get_task(task_id)
|
|
if task is None:
|
|
raise ValueError(f"Unknown task: {task_id}")
|
|
return task
|
|
|
|
async def _collect_related_tasks(self, task_id: str) -> list[Task]:
|
|
engine = await self.facade.ensure_ready()
|
|
if not engine.store:
|
|
return []
|
|
root_task = await self._get_task(task_id)
|
|
tasks = await engine.store.get_tasks(project_id=self._project_id)
|
|
related: list[Task] = []
|
|
root_session_id = str(root_task.session_id or "").strip()
|
|
for task in tasks:
|
|
metadata = task.metadata if isinstance(task.metadata, dict) else {}
|
|
origin_task_id = str(metadata.get("origin_task_id", "") or "").strip()
|
|
if task.id == root_task.id:
|
|
related.append(task)
|
|
continue
|
|
if origin_task_id and origin_task_id == root_task.id:
|
|
related.append(task)
|
|
continue
|
|
parent_session_id = str(task.parent_session_id or "").strip()
|
|
if root_session_id and parent_session_id and parent_session_id == root_session_id:
|
|
related.append(task)
|
|
return related
|
|
|
|
async def _mark_related_tasks(self, task_id: str, status: TaskStatus) -> None:
|
|
engine = await self.facade.ensure_ready()
|
|
if not engine.store:
|
|
return
|
|
for task in await self._collect_related_tasks(task_id):
|
|
await apply_task_status_transition(
|
|
engine.store,
|
|
task,
|
|
target_status_or_phase=status,
|
|
reason="cli_board_mark_related_tasks",
|
|
release_claim=status == TaskStatus.CANCELLED,
|
|
)
|
|
|
|
async def _resolve_related_checkpoints(self, task_id: str, *, status: str) -> None:
|
|
engine = await self.facade.ensure_ready()
|
|
if not engine.store:
|
|
return
|
|
|
|
related = await self._collect_related_tasks(task_id)
|
|
if not related:
|
|
return
|
|
|
|
task_ids = {task.id for task in related}
|
|
session_ids = {task.session_id for task in related if task.session_id}
|
|
checkpoints = await engine.store.get_pending_checkpoints(project_id=self._project_id)
|
|
for checkpoint in checkpoints:
|
|
if checkpoint.task_id in task_ids or checkpoint.session_id in session_ids:
|
|
await engine.store.resolve_execution_checkpoint(checkpoint.checkpoint_id, status=status)
|