From 48419e3f69cd0dc3e83288eb7e94e5eb29cd8d46 Mon Sep 17 00:00:00 2001 From: Charles Yao Date: Tue, 18 Aug 2026 11:14:48 -0700 Subject: [PATCH] feat(copilot): copilot session usecase + dispatch --- api/services/workflow_copilot/service.py | 163 ++++++++++++++ .../services/workflow_copilot/test_service.py | 199 ++++++++++++++++++ 2 files changed, 362 insertions(+) create mode 100644 api/services/workflow_copilot/service.py create mode 100644 api/tests/unit_tests/services/workflow_copilot/test_service.py diff --git a/api/services/workflow_copilot/service.py b/api/services/workflow_copilot/service.py new file mode 100644 index 00000000000..9ae9e8b3bc2 --- /dev/null +++ b/api/services/workflow_copilot/service.py @@ -0,0 +1,163 @@ +"""The Workflow Copilot usecase: session lifecycle + dispatch. + +Port of dify-enterprise/server/pkg/enterprise/biz/copilot/usecase.go. + +Dependency-injected (``repo``, ``session_lock``, ``enqueue_fn`` are +constructor args) so it stays unit-testable without importing the real +Redis-backed ``session_lock`` module or Celery -- those are wired in by the +caller (P3b Task 4: the Flask controller + the Celery task's ``.delay``). +""" + +from collections.abc import Callable +from dataclasses import dataclass +from typing import Protocol + +from core.workflow_copilot.errors import BusyError, ConflictError, NotFoundError +from core.workflow_copilot.models import ( + Action, + Actor, + ChecklistError, + ConversationItem, + EntryMode, + FixContext, + Session, +) +from core.workflow_copilot.ports import Repository +from core.workflow_copilot.state import PcState, canvas_read_only, is_waiting, is_working + +__all__ = ["SessionLock", "SessionView", "WorkflowCopilotService"] + + +@dataclass +class SessionView: + """Read model returned by the usecase's session-facing methods.""" + + session_id: str + app_id: str + version: int + state: str # str(PcState), e.g. "fix.await_verify" + canvas_read_only: bool + run_status: str + interrupted: bool + conversation: list[ConversationItem] + + +class SessionLock(Protocol): + """The per-session advance gate. Both the ``session_lock`` module and + the in-memory fake used in tests satisfy this structurally.""" + + def acquire(self, session_id: str) -> str | None: ... + + def release(self, session_id: str, token: str) -> None: ... + + def exists(self, session_id: str) -> bool: ... + + +def _run_status(state: PcState) -> str: + """Port of Go ``runStatusFor``.""" + if state == PcState.SUCCESS: + return "complete" + if state == PcState.FAILED: + return "failed" + if is_waiting(state): + return "waiting-input" + if is_working(state): + return "executing" + return "unknown" + + +class WorkflowCopilotService: + """Session lifecycle (create/view/action/message) + dispatch. + + Port of Go ``usecase.go``'s ``Usecase``. + """ + + def __init__( + self, + repo: Repository, + session_lock: SessionLock, + enqueue_fn: Callable[[str, Action, Actor, str], None], + ) -> None: + self._repo = repo + self._session_lock = session_lock + self._enqueue_fn = enqueue_fn + + def create_fix_session( + self, + app_id: str, + actor: Actor, + failed_run_id: str | None = None, + checklist_errors: list[ChecklistError] | None = None, + ) -> SessionView: + """Port of Go ``CreateFixSession``. + + Checklist takes precedence when errors are present. Does NOT create + a failed ``CopilotRun`` row -- ``failed_run_id`` is trusted to + already name a recorded run; recording it is an upstream/P4 + concern. + """ + if checklist_errors: + entry_mode, state = EntryMode.FIX_CHECKLIST, PcState.CHECKLIST_DIAGNOSE + fc = FixContext(source="checklist", checklist_errors=checklist_errors) + else: + entry_mode, state = EntryMode.FIX, PcState.FIX_DIAGNOSE + fc = FixContext(failed_run_id=failed_run_id or "", source="run") + + s = Session( + app_id=app_id, + tenant_id=actor.tenant_id, + owner_account_id=actor.account_id, + entry_mode=entry_mode, + current_state=state, + ) + items = [ConversationItem(kind="run-context", seq=0, payload={"run_id": failed_run_id or ""})] + self._repo.create_session(s, fc, items) # assigns s.id, s.version = 1 + + self.dispatch(s.id, Action(kind="request_fix", base_version=1), actor) + return self.get_session_view(s.id, actor) + + def get_session_view(self, session_id: str, actor: Actor) -> SessionView: + """Port of Go ``GetSessionView``.""" + s, _fc = self._repo.get_session(session_id) # raises NotFoundError if absent + if s.owner_account_id != actor.account_id: + raise NotFoundError("session not found") # do not leak existence to non-owners + items = self._repo.list_conversation(session_id) + st = s.current_state + return SessionView( + session_id=s.id, + app_id=s.app_id, + version=s.version, + state=str(st), + canvas_read_only=canvas_read_only(st), + run_status=_run_status(st), + interrupted=is_working(st) and not self._session_lock.exists(session_id), + conversation=items, + ) + + def submit_action(self, session_id: str, actor: Actor, action: Action) -> SessionView: + """Port of Go ``SubmitAction``.""" + s, _fc = self._repo.get_session(session_id) + if s.owner_account_id != actor.account_id: + raise NotFoundError("session not found") + if action.base_version != s.version: + raise ConflictError(f"stale base_version {action.base_version} for session {session_id}") + self.dispatch(session_id, action, actor) + return self.get_session_view(session_id, actor) + + def submit_message(self, session_id: str, actor: Actor, text: str, base_version: int) -> SessionView: + """Port of Go ``SubmitMessage``.""" + return self.submit_action( + session_id, actor, Action(kind="message", payload={"text": text}, base_version=base_version) + ) + + def dispatch(self, session_id: str, action: Action, actor: Actor) -> None: + """Port of Go ``dispatch``: acquire-or-busy + enqueue, release on + enqueue failure.""" + token = self._session_lock.acquire(session_id) + if token is None: + raise BusyError(f"session {session_id} is busy") + try: + self._enqueue_fn(session_id, action, actor, token) + except Exception: + self._session_lock.release(session_id, token) + raise diff --git a/api/tests/unit_tests/services/workflow_copilot/test_service.py b/api/tests/unit_tests/services/workflow_copilot/test_service.py new file mode 100644 index 00000000000..00b7acc75bb --- /dev/null +++ b/api/tests/unit_tests/services/workflow_copilot/test_service.py @@ -0,0 +1,199 @@ +"""Unit tests for the ``WorkflowCopilotService`` usecase (P3b Task 3). + +Wires ``SqlCopilotRepository`` (backed by a SQLite in-memory DB, same +fixtures as ``test_engine_on_sql_repo.py``) with a fake in-memory +``session_lock`` and a capturing ``enqueue_fn`` -- the service is +dependency-injected so it stays unit-testable without Celery or the real +Redis-backed ``session_lock`` module. +""" + +from collections.abc import Iterator + +import pytest +from sqlalchemy import create_engine +from sqlalchemy.engine import Engine +from sqlalchemy.orm import sessionmaker + +from core.workflow_copilot.errors import BusyError, ConflictError, NotFoundError +from core.workflow_copilot.models import Action, Actor, ChecklistError +from models.base import Base +from services.workflow_copilot.repository import SqlCopilotRepository +from services.workflow_copilot.service import WorkflowCopilotService + +TENANT_ID = "11111111-1111-1111-1111-111111111111" +APP_ID = "22222222-2222-2222-2222-222222222222" +ACCOUNT_ID = "33333333-3333-3333-3333-333333333333" +OTHER_ACCOUNT_ID = "44444444-4444-4444-4444-444444444444" + + +class FakeSessionLock: + """In-memory stand-in for the Redis-backed ``session_lock`` module.""" + + def __init__(self) -> None: + self._held: dict[str, str] = {} + + def acquire(self, session_id: str) -> str | None: + if session_id in self._held: + return None + token = f"tok-{session_id}" + self._held[session_id] = token + return token + + def release(self, session_id: str, token: str) -> None: + if self._held.get(session_id) == token: + self._held.pop(session_id, None) + + def exists(self, session_id: str) -> bool: + return session_id in self._held + + +@pytest.fixture +def engine() -> Iterator[Engine]: + engine = create_engine("sqlite://") + Base.metadata.create_all(engine) + try: + yield engine + finally: + engine.dispose() + + +@pytest.fixture +def repo(engine: Engine) -> SqlCopilotRepository: + factory = sessionmaker(bind=engine, expire_on_commit=False) + return SqlCopilotRepository(factory) + + +@pytest.fixture +def lock() -> FakeSessionLock: + return FakeSessionLock() + + +@pytest.fixture +def enqueued() -> list[tuple]: + return [] + + +@pytest.fixture +def service(repo: SqlCopilotRepository, lock: FakeSessionLock, enqueued: list[tuple]) -> WorkflowCopilotService: + def enqueue_fn(session_id, action, actor, token) -> None: + enqueued.append((session_id, action, actor, token)) + + return WorkflowCopilotService(repo=repo, session_lock=lock, enqueue_fn=enqueue_fn) + + +def _actor(account_id: str = ACCOUNT_ID) -> Actor: + return Actor(account_id=account_id, tenant_id=TENANT_ID) + + +def test_create_fix_session_dispatches_request_fix_and_holds_lock( + service: WorkflowCopilotService, lock: FakeSessionLock, enqueued: list[tuple] +) -> None: + actor = _actor() + + view = service.create_fix_session(APP_ID, actor, failed_run_id="TR-1") + + assert view.state == "fix.diagnose" + assert view.canvas_read_only is True + assert view.run_status == "executing" + assert view.app_id == APP_ID + assert view.version == 1 + + # the lock was acquired by dispatch and never released (enqueue_fn is a + # pure capture -- it does not run the task). + assert lock.exists(view.session_id) is True + + assert len(enqueued) == 1 + sid, action, dispatched_actor, token = enqueued[0] + assert sid == view.session_id + assert action == Action(kind="request_fix", base_version=1) + assert dispatched_actor == actor + assert token == f"tok-{view.session_id}" + + +def test_create_fix_session_checklist_entry(service: WorkflowCopilotService) -> None: + actor = _actor() + errors = [ChecklistError(node_id="n1", node_type="llm", title="LLM", messages=["missing prompt"])] + + view = service.create_fix_session(APP_ID, actor, checklist_errors=errors) + + assert view.state == "checklist.diagnose" + assert view.canvas_read_only is True + + +def test_submit_action_while_lock_held_raises_busy( + service: WorkflowCopilotService, enqueued: list[tuple] +) -> None: + actor = _actor() + view = service.create_fix_session(APP_ID, actor, failed_run_id="TR-1") + assert len(enqueued) == 1 + + # CAS passes (version is 1), but dispatch's acquire fails because the + # lock from create_fix_session's dispatch is still held. + with pytest.raises(BusyError): + service.submit_action(view.session_id, actor, Action(kind="run_verify", base_version=1)) + + # no new enqueue happened for the rejected action. + assert len(enqueued) == 1 + + +def test_submit_action_with_stale_base_version_raises_conflict( + service: WorkflowCopilotService, enqueued: list[tuple] +) -> None: + actor = _actor() + view = service.create_fix_session(APP_ID, actor, failed_run_id="TR-1") + assert len(enqueued) == 1 + + with pytest.raises(ConflictError): + service.submit_action(view.session_id, actor, Action(kind="run_verify", base_version=999)) + + # the CAS check happens before dispatch -- no new enqueue. + assert len(enqueued) == 1 + + +def test_get_session_view_by_non_owner_raises_not_found(service: WorkflowCopilotService) -> None: + actor = _actor() + other = _actor(OTHER_ACCOUNT_ID) + view = service.create_fix_session(APP_ID, actor, failed_run_id="TR-1") + + with pytest.raises(NotFoundError): + service.get_session_view(view.session_id, other) + + +def test_submit_action_by_non_owner_raises_not_found(service: WorkflowCopilotService) -> None: + actor = _actor() + other = _actor(OTHER_ACCOUNT_ID) + view = service.create_fix_session(APP_ID, actor, failed_run_id="TR-1") + + with pytest.raises(NotFoundError): + service.submit_action(view.session_id, other, Action(kind="run_verify", base_version=1)) + + +def test_get_session_view_interrupted_reflects_lock_absence( + service: WorkflowCopilotService, lock: FakeSessionLock, enqueued: list[tuple] +) -> None: + actor = _actor() + view = service.create_fix_session(APP_ID, actor, failed_run_id="TR-1") + sid, _action, _actor2, token = enqueued[0] + + # lock is held (dispatch's acquire succeeded) -- fix.diagnose is a + # working state, so with the lock held it must NOT be interrupted. + view = service.get_session_view(sid, actor) + assert view.state == "fix.diagnose" + assert view.interrupted is False + + # release the lock out-of-band (simulating a crashed/expired worker) -- + # the working state now reads as interrupted. + lock.release(sid, token) + view = service.get_session_view(sid, actor) + assert view.interrupted is True + + +def test_submit_message_wraps_submit_action(service: WorkflowCopilotService) -> None: + actor = _actor() + view = service.create_fix_session(APP_ID, actor, failed_run_id="TR-1") + + # lock is held from create_fix_session's dispatch, so this must raise + # BusyError -- proving submit_message routes through submit_action -> + # dispatch just like a plain action would. + with pytest.raises(BusyError): + service.submit_message(view.session_id, actor, "hello", base_version=1)