mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-01 19:06:01 +00:00
c38d291505
Replaces the pre-hexagonal scheduled-task implementation with a slice built to the layering spec: a pure domain (two aggregates, two state machines, the policy value object), output ports it declares itself, SQL/launcher/thread adapters implementing them under `app/adapters/`, and a composition root that is the one place any of them is instantiated. The old implementation mixed all of that into `app/scheduler/service.py` and a router that reached straight into repositories, so the rules that matter -- overlap policy, lease handling, which write owns which timestamp -- were only reachable through a live database. They are now unit-assertable on in-memory fakes, with the contract suite running each port against both the fake and real sqlite, and the concurrency invariants pinned by dedicated race tests. Two bugs the old shape hid are fixed on the way: a completion hook that replayed a stale snapshot and rolled back the launch write, and a corrupt stored row surfacing to the client as a 4xx. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
90 lines
3.9 KiB
Python
90 lines
3.9 KiB
Python
"""Secondary adapter (anti-corruption layer) -- starting a run through Gateway.
|
|
|
|
Implements ``RunLauncher`` from ``deerflow.domain.schedule.ports``. This context
|
|
owns no part of the run lifecycle: it asks the Gateway to start one and
|
|
translates whatever comes back into the two outcomes the domain distinguishes.
|
|
|
|
That translation is the reason this file exists. The Gateway signals a busy
|
|
thread two different ways -- ``ConflictError`` from the run manager, or an
|
|
``HTTPException(409)`` from the route-level path -- and the legacy scheduler
|
|
service therefore imported ``fastapi`` to tell them apart. Both are the same
|
|
domain fact, and saying so here is what keeps the web framework and the run
|
|
runtime out of the inner ring.
|
|
|
|
TODO(hexagonal): this depends on ``launch_scheduled_thread_run``, a Gateway
|
|
service function returning an untyped dict, rather than on a contract published
|
|
by the run context -- that context has not been through a hexagonal slice yet.
|
|
When it publishes one (a DTO, not its aggregate and not its repository),
|
|
replace the body of this class. The ``RunLauncher`` port does not move.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Awaitable, Callable, Mapping
|
|
from typing import Any
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from deerflow.domain.schedule.exceptions import LaunchFailedError, ThreadBusyError
|
|
from deerflow.domain.schedule.ports import LaunchedRun, RunLauncher
|
|
from deerflow.runtime import ConflictError
|
|
|
|
LaunchRun = Callable[..., Awaitable[Mapping[str, Any]]]
|
|
|
|
|
|
class GatewayRunLauncher(RunLauncher):
|
|
"""Adapts the Gateway's scheduled-run launch path to the ``RunLauncher`` port.
|
|
|
|
Takes the launch callable rather than importing it, because the production
|
|
one is bound to the FastAPI app (``launch_scheduled_thread_run(app=app,
|
|
...)``) and that binding belongs to the composition root.
|
|
|
|
Explicit inheritance is a readability aid only: a misspelled method would
|
|
still instantiate fine and silently inherit the Protocol's ``...`` body,
|
|
so the contract tests must call every port method and assert on what it
|
|
returns.
|
|
"""
|
|
|
|
def __init__(self, launch_run: LaunchRun) -> None:
|
|
self._launch_run = launch_run
|
|
|
|
async def launch(
|
|
self,
|
|
*,
|
|
thread_id: str,
|
|
assistant_id: str | None,
|
|
prompt: str,
|
|
owner_user_id: str | None,
|
|
metadata: dict[str, str],
|
|
) -> LaunchedRun:
|
|
try:
|
|
result = await self._launch_run(
|
|
thread_id=thread_id,
|
|
assistant_id=assistant_id,
|
|
prompt=prompt,
|
|
owner_user_id=owner_user_id,
|
|
metadata=metadata,
|
|
)
|
|
except ConflictError as exc:
|
|
raise ThreadBusyError(str(exc)) from exc
|
|
except HTTPException as exc:
|
|
if exc.status_code == 409:
|
|
raise ThreadBusyError(str(exc.detail)) from exc
|
|
raise LaunchFailedError(str(exc.detail)) from exc
|
|
except Exception as exc:
|
|
# Deliberately broad: the port promises the domain that nothing but
|
|
# its two errors escapes, so an unclassifiable failure has to become
|
|
# the "genuine failure" branch rather than unwinding the poll loop.
|
|
# `CancelledError` derives from BaseException and is not caught --
|
|
# shutdown is control flow, not a launch outcome.
|
|
raise LaunchFailedError(str(exc)) from exc
|
|
|
|
run_id = result.get("run_id")
|
|
launched_thread_id = result.get("thread_id")
|
|
if not isinstance(run_id, str) or not isinstance(launched_thread_id, str):
|
|
# The run path broke its own contract. Reporting it as a failure
|
|
# keeps the task's bookkeeping honest instead of recording a launch
|
|
# whose run can never be traced.
|
|
raise LaunchFailedError(f"run launch returned no usable identity: {result!r}")
|
|
return LaunchedRun(run_id=run_id, thread_id=launched_thread_id)
|