diff --git a/backend/app/adapters/schedule/__init__.py b/backend/app/adapters/schedule/__init__.py index e69de29bb..bc75798ac 100644 --- a/backend/app/adapters/schedule/__init__.py +++ b/backend/app/adapters/schedule/__init__.py @@ -0,0 +1,20 @@ +"""Adapters of the schedule context. + +Mostly secondary (driven) adapters -- implementations of the output ports the +domain declares, which the service calls out to: + + scheduled_task_repository.py owned persistence + scheduled_run_repository.py owned persistence + run_launcher.py anti-corruption layer over the run runtime + thread_lookup.py anti-corruption layer over the thread store + +One exception, and the name says so: + + run_completion.py PRIMARY (inbound) -- the run runtime calls it + +It lives here so the context stays in one place rather than beside the other +two primary adapters, which sit next to whatever drives them (the router under +`gateway/routers/schedule/`, the poller under `scheduler/`). Direction is +stated by each module's own first line; a file added here without one is a +file whose direction nobody decided. +""" diff --git a/backend/app/adapters/schedule/run_completion.py b/backend/app/adapters/schedule/run_completion.py new file mode 100644 index 000000000..83007371e --- /dev/null +++ b/backend/app/adapters/schedule/run_completion.py @@ -0,0 +1,102 @@ +"""Primary adapter (inbound) -- the run runtime's completion callback. + +The one file in this package that drives the domain rather than serving it. +Its siblings are secondary adapters the service calls out to; this one is +called *by* the run runtime, the same way the router is called by HTTP and the +poller by its clock. Kept here rather than beside those two so the context +stays in one place, with the direction stated by the name and by this line. + +Its job is the filtering the legacy completion hook did inline. Every run in +the process reaches that callback, so most of them are none of this context's +business, and producing no call at all says exactly that -- not an error, just +nothing to write back. That is why `ScheduleService.handle_run_completion` +carries no guard clauses and never imports ``RunRecord``. + +TODO(hexagonal): this depends on ``RunRecord``, a run-runtime type, 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 ``_to_outcome``. Nothing else moves. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import TYPE_CHECKING + +from deerflow.domain.schedule.model import RunStatus +from deerflow.domain.schedule.ports import RunOutcome + +if TYPE_CHECKING: + from deerflow.domain.schedule.service import ScheduleService + from deerflow.runtime import RunRecord + +# The runtime reports four terminal states; the domain has three, because +# `timeout` and `error` are the same fact to a scheduled task while +# `interrupted` is deliberately not -- a cancel or same-thread takeover ends +# the task CANCELLED, not FAILED. +_TERMINAL_STATUSES = { + "success": RunStatus.SUCCESS, + "error": RunStatus.FAILED, + "timeout": RunStatus.FAILED, + "interrupted": RunStatus.INTERRUPTED, +} + +_INTERRUPTED_WITHOUT_ERROR = "run was interrupted before completion" + + +class ScheduleRunCompletionListener: + """Turns a finished run into the write-back use case, or into nothing. + + Deciding whether a run is ours and invoking the use case are one + responsibility, not two: "ignore this run" is only meaningful as "do not + call the service", so splitting them left the second half living in the + composition root, where behaviour is not asserted. + """ + + def __init__(self, service: ScheduleService) -> None: + self._service = service + + async def __call__(self, record: RunRecord) -> None: + outcome = self._to_outcome(record) + if outcome is None: + return + await self._service.handle_run_completion(outcome, now=datetime.now(UTC)) + + @staticmethod + def _to_outcome(record: RunRecord) -> RunOutcome | None: + """Translate into domain vocabulary, or `None` to ignore the run. + + `None` when the run is not a scheduled execution (no usable task + metadata, no owner) or has not reached a terminal state yet. + """ + metadata = record.metadata or {} + task_id = metadata.get("scheduled_task_id") + record_id = metadata.get("scheduled_task_run_id") + user_id = record.user_id + # `metadata` is a free-form dict a caller can influence, so the ids are + # type-checked rather than assumed; `user_id` is required because every + # task read is scoped by it. + if not isinstance(task_id, str) or not isinstance(record_id, str) or not user_id: + return None + + status = _TERMINAL_STATUSES.get(str(record.status.value)) + if status is None: + return None + + if status is RunStatus.SUCCESS: + # A stale error left on a successful record must not be written back + # as the task's last_error. + error = None + elif status is RunStatus.INTERRUPTED: + error = record.error or _INTERRUPTED_WITHOUT_ERROR + else: + error = record.error + + return RunOutcome( + task_id=task_id, + record_id=record_id, + run_id=record.run_id, + user_id=user_id, + status=status, + error=error, + ) diff --git a/backend/app/adapters/schedule/run_outcome_mapping.py b/backend/app/adapters/schedule/run_outcome_mapping.py deleted file mode 100644 index 40f0e1b44..000000000 --- a/backend/app/adapters/schedule/run_outcome_mapping.py +++ /dev/null @@ -1,74 +0,0 @@ -"""Boundary mapping (not a port implementation) -- RunRecord -> RunOutcome. - -Unlike its siblings in this package, this module implements no port: it is the -inbound translation the composition root installs on the run runtime's -completion hook, so the domain never imports ``RunRecord``. - -It also owns the filtering the legacy completion hook did inline. Every run in -the process reaches that hook, so most of them are none of this context's -business, and returning ``None`` says exactly that -- not an error, just -nothing to write back. The service is then never called at all, which is why it -has no guard clauses of its own. -""" - -from __future__ import annotations - -from typing import TYPE_CHECKING - -from deerflow.domain.schedule.model import RunStatus -from deerflow.domain.schedule.ports import RunOutcome - -if TYPE_CHECKING: - from deerflow.runtime import RunRecord - -# The runtime reports four terminal states; the domain has three, because -# `timeout` and `error` are the same fact to a scheduled task while -# `interrupted` is deliberately not -- a cancel or same-thread takeover ends -# the task CANCELLED, not FAILED. -_TERMINAL_STATUSES = { - "success": RunStatus.SUCCESS, - "error": RunStatus.FAILED, - "timeout": RunStatus.FAILED, - "interrupted": RunStatus.INTERRUPTED, -} - -_INTERRUPTED_WITHOUT_ERROR = "run was interrupted before completion" - - -def run_outcome_from_record(record: RunRecord) -> RunOutcome | None: - """Translate a finished run into domain vocabulary, or `None` to ignore it. - - `None` is returned when the run is not a scheduled execution (no usable - task metadata, no owner) or has not reached a terminal state yet. - """ - metadata = record.metadata or {} - task_id = metadata.get("scheduled_task_id") - record_id = metadata.get("scheduled_task_run_id") - user_id = record.user_id - # `metadata` is a free-form dict a caller can influence, so the ids are - # type-checked rather than assumed; `user_id` is required because every - # task read is scoped by it. - if not isinstance(task_id, str) or not isinstance(record_id, str) or not user_id: - return None - - status = _TERMINAL_STATUSES.get(str(record.status.value)) - if status is None: - return None - - if status is RunStatus.SUCCESS: - # A stale error left on a successful record must not be written back as - # the task's last_error. - error = None - elif status is RunStatus.INTERRUPTED: - error = record.error or _INTERRUPTED_WITHOUT_ERROR - else: - error = record.error - - return RunOutcome( - task_id=task_id, - record_id=record_id, - run_id=record.run_id, - user_id=user_id, - status=status, - error=error, - ) diff --git a/backend/app/composition.py b/backend/app/composition.py index c2a95cf74..48f876d1d 100644 --- a/backend/app/composition.py +++ b/backend/app/composition.py @@ -26,7 +26,6 @@ from __future__ import annotations from collections.abc import Awaitable, Callable, Mapping from dataclasses import dataclass -from datetime import UTC, datetime from typing import TYPE_CHECKING, Any from deerflow.domain.feedback.service import FeedbackService @@ -98,11 +97,10 @@ def build_run_completion_hook( ) -> Callable[[Any], Awaitable[None]] | None: """Install the schedule context on the run runtime's completion callback. - This is the inbound half of the wiring: the run runtime hands every - finished run to one callback, and `run_outcome_from_record` decides which - of them the schedule context has any business with. Runs that are not - scheduled executions produce no outcome and the service is never called, - which is why it carries no guard clauses of its own. + The inbound half of the wiring, and nothing more: which runs the context + cares about and what it does with them belongs to the adapter, not here. + This function's only decision is the same one the rest of this module + makes -- what to assemble, and what to leave unassembled. Returns ``None`` when there is no service, so the runtime installs no hook at all rather than one that always declines. @@ -110,14 +108,9 @@ def build_run_completion_hook( if schedule_service is None: return None - from app.adapters.schedule.run_outcome_mapping import run_outcome_from_record + from app.adapters.schedule.run_completion import ScheduleRunCompletionListener - async def on_run_completed(record: Any) -> None: - outcome = run_outcome_from_record(record) - if outcome is not None: - await schedule_service.handle_run_completion(outcome, now=datetime.now(UTC)) - - return on_run_completed + return ScheduleRunCompletionListener(schedule_service) def build_schedule_policy(scheduler_config: SchedulerConfig) -> SchedulePolicy: diff --git a/backend/docs/SCHEDULE_DESIGN_zh.md b/backend/docs/SCHEDULE_DESIGN_zh.md index 2541cdd60..057683d75 100644 --- a/backend/docs/SCHEDULE_DESIGN_zh.md +++ b/backend/docs/SCHEDULE_DESIGN_zh.md @@ -57,9 +57,10 @@ flowchart LR subgraph PRIM["✅ 主适配器 · app"] R["gateway/routers/schedule/
router · models"] PL["scheduler/poller.py
轮询时钟"] + RC["adapters/schedule/run_completion.py
运行完成回调"] end subgraph SEC["✅ 从适配器 · app"] - AD["adapters/schedule/
两个仓储 · run_launcher · thread_lookup
run_outcome_mapping"] + AD["adapters/schedule/
两个仓储 · run_launcher · thread_lookup"] end subgraph CR["✅ 组合根"] CO["composition.py
build_domain_services()"] @@ -630,6 +631,7 @@ await service.update_task( | [`gateway/routers/schedule/router.py`](../app/gateway/routers/schedule/router.py) | 10 个 HTTP 端点，只做协议转换 + 领域错误→状态码 | | [`gateway/routers/schedule/models.py`](../app/gateway/routers/schedule/models.py) | 请求/响应模型；响应是白名单，不是 ORM 转储；`schedule_spec` 的进出转换是模型自己的方法 | | [`app/scheduler/poller.py`](../app/scheduler/poller.py) | 轮询时钟 + 启动恢复 | +| [`adapters/schedule/run_completion.py`](../app/adapters/schedule/run_completion.py) | 运行完成回调；过滤掉非本上下文的运行，其余转成 `RunOutcome` 并调用用例。与上面两个同为入站，只是留在上下文包内 | **从适配器** @@ -639,7 +641,6 @@ await service.update_task( | [`adapters/schedule/scheduled_run_repository.py`](../app/adapters/schedule/scheduled_run_repository.py) | 自有持久化；`IntegrityError → ActiveRunConflictError` 的翻译点 | | [`adapters/schedule/run_launcher.py`](../app/adapters/schedule/run_launcher.py) | 防腐层；`ConflictError` / `HTTPException(409)` → `ThreadBusyError` | | [`adapters/schedule/thread_lookup.py`](../app/adapters/schedule/thread_lookup.py) | 防腐层；`check_access(require_existing=True)` | -| [`adapters/schedule/run_outcome_mapping.py`](../app/adapters/schedule/run_outcome_mapping.py) | `RunRecord → RunOutcome \| None`，承接旧完成钩子的内联过滤 | **组合根** diff --git a/backend/tests/test_composition.py b/backend/tests/test_composition.py index 554e11e49..3f6a55cc9 100644 --- a/backend/tests/test_composition.py +++ b/backend/tests/test_composition.py @@ -10,7 +10,7 @@ from __future__ import annotations import pytest -from app.composition import DomainServices, build_domain_services, build_schedule_policy +from app.composition import DomainServices, build_domain_services, build_run_completion_hook, build_schedule_policy from deerflow.config.scheduler_config import SchedulerConfig from deerflow.domain.schedule.model import SchedulePolicy @@ -139,3 +139,30 @@ class TestEveryPortGetsTheRightAdapter: def test_the_policy_reaches_the_service(self): service = _build(object(), SchedulerConfig(max_concurrent_runs=9)).schedule assert service._policy.max_concurrent_runs == 9 + + +class TestRunCompletionHook: + """The inbound half of the wiring. + + What the hook *does* with a run is asserted in + `test_schedule_run_completion.py`; all that is left here is the assembly + decision, which is this module's whole job. + """ + + def test_no_service_installs_no_hook(self): + """`None` rather than a hook that always declines: with no service + there is nothing for the runtime to call, and saying so lets it skip + the callback entirely.""" + assert build_run_completion_hook(None) is None + + def test_a_service_is_wrapped_in_the_inbound_adapter(self): + from app.adapters.schedule.run_completion import ScheduleRunCompletionListener + + hook = build_run_completion_hook(_build(object()).schedule) + assert isinstance(hook, ScheduleRunCompletionListener) + + def test_the_hook_is_bound_to_the_service_it_was_given(self): + """Deliberately white-box, for the same reason as the port assertions + above: a hook wired to the wrong service type-checks cleanly.""" + service = _build(object()).schedule + assert build_run_completion_hook(service)._service is service diff --git a/backend/tests/test_schedule_run_completion.py b/backend/tests/test_schedule_run_completion.py new file mode 100644 index 000000000..c029801b5 --- /dev/null +++ b/backend/tests/test_schedule_run_completion.py @@ -0,0 +1,176 @@ +"""Tests for the run-completion driving adapter. + +Every run in the process reaches the run runtime's completion callback, so +this adapter's first job is deciding which of them the schedule context has +any business with -- the filtering the legacy hook did inline, with four guard +clauses and an `if/elif` chain over runtime strings. A run that is not a +scheduled execution produces no call at all, which is why the service carries +no guard clauses of its own and never imports the run runtime. + +Driven through `__call__` rather than a conversion function: "was the use case +invoked, and with what" is the behaviour, and the previous split -- a pure +converter here plus the invoking closure in the composition root -- left that +second half untested, since the composition root is not where behaviour is +asserted. +""" + +from __future__ import annotations + +from datetime import UTC, datetime + +import pytest + +from app.adapters.schedule.run_completion import ScheduleRunCompletionListener +from deerflow.domain.schedule.model import RunStatus +from deerflow.domain.schedule.ports import RunOutcome +from deerflow.runtime.runs.manager import RunRecord +from deerflow.runtime.runs.schemas import DisconnectMode +from deerflow.runtime.runs.schemas import RunStatus as RuntimeRunStatus + +pytestmark = pytest.mark.asyncio + +TASK_METADATA = { + "scheduled_task_id": "task-1", + "scheduled_task_run_id": "rec-1", + "scheduled_trigger": "scheduled", +} + +_DEFAULT = object() + + +def _record(*, status, metadata=_DEFAULT, user_id="user-1", error=None, run_id="run-1"): + return RunRecord( + run_id=run_id, + thread_id="thread-1", + assistant_id=None, + status=status, + on_disconnect=DisconnectMode.continue_, + metadata=TASK_METADATA if metadata is _DEFAULT else metadata, + user_id=user_id, + error=error, + ) + + +class _RecordingService: + """Stands in for `ScheduleService`, capturing what the use case received.""" + + def __init__(self) -> None: + self.calls: list[tuple[RunOutcome, datetime]] = [] + + async def handle_run_completion(self, outcome: RunOutcome, *, now: datetime) -> None: + self.calls.append((outcome, now)) + + +async def _delivered(record) -> RunOutcome | None: + """The outcome the service was called with, or None when it was not.""" + service = _RecordingService() + await ScheduleRunCompletionListener(service)(record) + return service.calls[0][0] if service.calls else None + + +class TestTerminalStatusMapping: + async def test_success(self): + assert await _delivered(_record(status=RuntimeRunStatus.success)) == RunOutcome( + task_id="task-1", + record_id="rec-1", + run_id="run-1", + user_id="user-1", + status=RunStatus.SUCCESS, + error=None, + ) + + async def test_a_successful_run_carries_no_error_even_if_the_record_has_one(self): + """A stale `error` on a successful record must not be written back as + the task's `last_error`; success is success.""" + outcome = await _delivered(_record(status=RuntimeRunStatus.success, error="stale")) + assert outcome is not None and outcome.error is None + + @pytest.mark.parametrize("status", [RuntimeRunStatus.error, RuntimeRunStatus.timeout]) + async def test_error_and_timeout_both_become_failed(self, status): + outcome = await _delivered(_record(status=status, error="boom")) + assert outcome is not None + assert outcome.status is RunStatus.FAILED + assert outcome.error == "boom" + + async def test_interrupted_is_distinct_from_failed(self): + """Red line: a cancel or same-thread takeover is not an execution + failure -- the task must end CANCELLED, and only INTERRUPTED gets it + there.""" + outcome = await _delivered(_record(status=RuntimeRunStatus.interrupted)) + assert outcome is not None and outcome.status is RunStatus.INTERRUPTED + + async def test_an_interrupt_without_an_error_gets_an_explanatory_one(self): + outcome = await _delivered(_record(status=RuntimeRunStatus.interrupted, error=None)) + assert outcome is not None and outcome.error == "run was interrupted before completion" + + async def test_an_interrupts_own_error_is_preserved(self): + outcome = await _delivered(_record(status=RuntimeRunStatus.interrupted, error="cancelled by user")) + assert outcome is not None and outcome.error == "cancelled by user" + + +class TestFilteredOut: + """None of these reach the service at all -- not as an error, not as a + no-op call it has to recognise.""" + + @pytest.mark.parametrize("status", [RuntimeRunStatus.pending, RuntimeRunStatus.running]) + async def test_a_non_terminal_run_is_ignored(self, status): + assert await _delivered(_record(status=status)) is None + + async def test_an_ordinary_chat_run_is_ignored(self): + """Every run in the process reaches this callback; only scheduled + executions carry the metadata that makes one ours.""" + assert await _delivered(_record(status=RuntimeRunStatus.success, metadata={})) is None + + @pytest.mark.parametrize( + "metadata", + [ + {"scheduled_task_run_id": "rec-1"}, + {"scheduled_task_id": "task-1"}, + {"scheduled_task_id": "task-1", "scheduled_task_run_id": None}, + {"scheduled_task_id": 7, "scheduled_task_run_id": "rec-1"}, + {"scheduled_task_id": "task-1", "scheduled_task_run_id": 7}, + ], + ) + async def test_half_or_mistyped_metadata_is_ignored(self, metadata): + """Both ids are required and both must be strings -- `metadata` is a + free-form dict a client can influence, so its shape is not assumed.""" + assert await _delivered(_record(status=RuntimeRunStatus.success, metadata=metadata)) is None + + @pytest.mark.parametrize("user_id", [None, ""]) + async def test_a_run_without_an_owner_is_ignored(self, user_id): + """Every task read is scoped by user_id; without one there is no + authorized way to look the task up.""" + assert await _delivered(_record(status=RuntimeRunStatus.success, user_id=user_id)) is None + + async def test_none_metadata_is_ignored(self): + """`metadata` is typed as a dict but the legacy hook defended against + `None`; keep that defence rather than rediscovering it in production.""" + assert await _delivered(_record(status=RuntimeRunStatus.success, metadata=None)) is None + + async def test_the_service_is_left_entirely_alone(self): + """Explicitly: not called, rather than called with something falsy. + + This half had no coverage while the invoking closure lived in the + composition root. + """ + service = _RecordingService() + await ScheduleRunCompletionListener(service)(_record(status=RuntimeRunStatus.running)) + assert service.calls == [] + + +class TestUseCaseInvocation: + async def test_the_completion_is_stamped_with_the_current_instant(self): + service = _RecordingService() + before = datetime.now(UTC) + await ScheduleRunCompletionListener(service)(_record(status=RuntimeRunStatus.success)) + after = datetime.now(UTC) + + assert len(service.calls) == 1 + _, now = service.calls[0] + assert now.tzinfo is not None, "the domain is tz-aware throughout" + assert before <= now <= after + + async def test_one_finished_run_produces_exactly_one_call(self): + service = _RecordingService() + await ScheduleRunCompletionListener(service)(_record(status=RuntimeRunStatus.success)) + assert len(service.calls) == 1 diff --git a/backend/tests/test_schedule_run_outcome_mapping.py b/backend/tests/test_schedule_run_outcome_mapping.py deleted file mode 100644 index 65e8152da..000000000 --- a/backend/tests/test_schedule_run_outcome_mapping.py +++ /dev/null @@ -1,122 +0,0 @@ -"""Boundary tests for the RunRecord -> RunOutcome conversion. - -This converter is where the completion hook's inline filtering moved to. The -old `handle_run_completion` opened with four guard clauses and a status -`if/elif` chain over runtime strings; the service now receives a `RunOutcome` -or is never called at all, so all of that lives here and the domain never -imports the run runtime. - -`None` therefore has one meaning: *this run is none of the schedule context's -business*. It is not an error and must never reach the service. -""" - -from __future__ import annotations - -import pytest - -from app.adapters.schedule.run_outcome_mapping import run_outcome_from_record -from deerflow.domain.schedule.model import RunStatus -from deerflow.domain.schedule.ports import RunOutcome -from deerflow.runtime.runs.manager import RunRecord -from deerflow.runtime.runs.schemas import DisconnectMode -from deerflow.runtime.runs.schemas import RunStatus as RuntimeRunStatus - -TASK_METADATA = { - "scheduled_task_id": "task-1", - "scheduled_task_run_id": "rec-1", - "scheduled_trigger": "scheduled", -} - - -_DEFAULT = object() - - -def _record(*, status, metadata=_DEFAULT, user_id="user-1", error=None, run_id="run-1"): - return RunRecord( - run_id=run_id, - thread_id="thread-1", - assistant_id=None, - status=status, - on_disconnect=DisconnectMode.continue_, - metadata=TASK_METADATA if metadata is _DEFAULT else metadata, - user_id=user_id, - error=error, - ) - - -class TestTerminalStatusMapping: - def test_success(self): - outcome = run_outcome_from_record(_record(status=RuntimeRunStatus.success)) - assert outcome == RunOutcome( - task_id="task-1", - record_id="rec-1", - run_id="run-1", - user_id="user-1", - status=RunStatus.SUCCESS, - error=None, - ) - - def test_a_successful_run_carries_no_error_even_if_the_record_has_one(self): - """A stale `error` on a successful record must not be written back as - the task's `last_error`; success is success.""" - outcome = run_outcome_from_record(_record(status=RuntimeRunStatus.success, error="stale")) - assert outcome is not None and outcome.error is None - - @pytest.mark.parametrize("status", [RuntimeRunStatus.error, RuntimeRunStatus.timeout]) - def test_error_and_timeout_both_become_failed(self, status): - outcome = run_outcome_from_record(_record(status=status, error="boom")) - assert outcome is not None - assert outcome.status is RunStatus.FAILED - assert outcome.error == "boom" - - def test_interrupted_is_distinct_from_failed(self): - """Red line: a cancel or same-thread takeover is not an execution - failure -- the task must end CANCELLED, and only INTERRUPTED gets it - there.""" - outcome = run_outcome_from_record(_record(status=RuntimeRunStatus.interrupted)) - assert outcome is not None and outcome.status is RunStatus.INTERRUPTED - - def test_an_interrupt_without_an_error_gets_an_explanatory_one(self): - outcome = run_outcome_from_record(_record(status=RuntimeRunStatus.interrupted, error=None)) - assert outcome is not None and outcome.error == "run was interrupted before completion" - - def test_an_interrupts_own_error_is_preserved(self): - outcome = run_outcome_from_record(_record(status=RuntimeRunStatus.interrupted, error="cancelled by user")) - assert outcome is not None and outcome.error == "cancelled by user" - - -class TestFilteredOut: - @pytest.mark.parametrize("status", [RuntimeRunStatus.pending, RuntimeRunStatus.running]) - def test_a_non_terminal_run_produces_nothing(self, status): - assert run_outcome_from_record(_record(status=status)) is None - - def test_a_run_with_no_metadata_produces_nothing(self): - """An ordinary chat run reaches the same completion hook; it is not a - scheduled execution and must not be written back as one.""" - assert run_outcome_from_record(_record(status=RuntimeRunStatus.success, metadata={})) is None - - @pytest.mark.parametrize( - "metadata", - [ - {"scheduled_task_run_id": "rec-1"}, - {"scheduled_task_id": "task-1"}, - {"scheduled_task_id": "task-1", "scheduled_task_run_id": None}, - {"scheduled_task_id": 7, "scheduled_task_run_id": "rec-1"}, - {"scheduled_task_id": "task-1", "scheduled_task_run_id": 7}, - ], - ) - def test_half_or_mistyped_metadata_produces_nothing(self, metadata): - """Both ids are required and both must be strings -- `metadata` is a - free-form dict a client can influence, so its shape is not assumed.""" - assert run_outcome_from_record(_record(status=RuntimeRunStatus.success, metadata=metadata)) is None - - @pytest.mark.parametrize("user_id", [None, ""]) - def test_a_run_without_an_owner_produces_nothing(self, user_id): - """Every task read is scoped by user_id; without one there is no - authorized way to look the task up.""" - assert run_outcome_from_record(_record(status=RuntimeRunStatus.success, user_id=user_id)) is None - - def test_a_none_metadata_produces_nothing(self): - """`metadata` is typed as a dict but the legacy hook defended against - `None`; keep that defence rather than rediscovering it in production.""" - assert run_outcome_from_record(_record(status=RuntimeRunStatus.success, metadata=None)) is None