mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-01 19:06:01 +00:00
9 Commits
| Author | SHA1 | Message | Date | |
|---|---|---|---|---|
rayhpeng
|
a6388bc967 |
fix(schedule): give each bookkeeping timestamp a single owner
created_at was minted twice for one construction: the aggregate's field default read the clock, then the adapter's add() read it again and stored its own value -- so the aggregate a caller held disagreed with the stored row by however long the insert took, and the model docstring pointed at a write path that no longer exists. ScheduledTask.create now stamps created_at and updated_at from the explicit now= it already receives (one clock reading, the rule input), and the adapter persists the aggregate's instants verbatim on insert -- matching the feedback reference implementation. updated_at's later life stays deliberately with the adapter's write paths: the CAS methods (record_launch / record_completion) write the row without ever holding an aggregate, so only storage can stamp 'last written'. The model docstring now states that ownership split instead of citing the deleted legacy repository. Pinned by a domain test on the factory and a contract case on add() across both implementations. |
||
|
rayhpeng
|
06dda5045a |
fix(schedule): stop a corrupt stored row surfacing as a client error
SqlScheduledTaskRepository._to_domain raised InvalidScheduleError for a row whose stored schedule no longer parses -- the same error the aggregate raises for a client-submitted schedule, which the router maps to 422. A stored fault therefore told the client its perfectly fine request was wrong, and made the row unrepairable over HTTP: PATCH reads the task before writing, so the fix path 422'd too. Enum rebuild failures were worse -- a raw ValueError crossed the boundary untranslated. The rebuild is now translated to a dedicated CorruptStoredScheduleError, raised only by the persistence adapter and deliberately absent from the router's status table, so it falls through to the unclassified-500 branch: a server-side fault reported as one. List reads keep skipping and logging the bad row. Pinned by tests/test_schedule_corrupt_rows.py against a real sqlite database. |
||
|
rayhpeng
|
f88c8e61bc |
refactor(schedule): promote the domain errors to exceptions.py
Move model/errors.py up one level to domain/schedule/exceptions.py, a sibling of the model, matching the AWS domain layout the spec mandates (exceptions/ is its own member of the domain folder, not part of the model) and the feedback reference implementation. Class names keep the PEP 8 Error suffix. Pure move -- the nine classes are AST-identical to the originals; imports across the domain, adapters, router, and tests now take errors from deerflow.domain.schedule.exceptions. |
||
|
rayhpeng
|
34ce7c259b |
refactor(schedule): make the run-completion hook an inbound adapter
`run_outcome_mapping.py` called itself "not a port implementation" and sat in a package of secondary adapters, while the half that actually invoked the use case lived as a closure in the composition root. It is one thing, and it is a primary adapter: the run runtime calls it the way HTTP calls the router and the clock calls the poller. `ScheduleRunCompletionListener` now holds the whole responsibility -- decide whether a finished run is ours, translate it, invoke the use case. Those are not two jobs: "ignore this run" is only meaningful as "do not call the service", so splitting them is what left the second half in a place where behaviour is not asserted. `build_run_completion_hook` drops to `return ScheduleRunCompletionListener(service)`. The composition root's own docstring says no adapter logic lives there; that is now true of it as well as of the routers it was written about. Placement --------- Kept in `app/adapters/schedule/` rather than moved beside the other two primary adapters. The context stays in one package; direction is stated by the class name and each module's first line, and the package `__init__` -- previously empty -- now lists which of its modules point which way, so a file added without that line is visibly a file whose direction nobody decided. A subdirectory for a single inbound module would have made the other four look like they had been sorted into something. Tests ----- This is the part that was not a rename. The conversion had 24 cases; the invocation had none, because the composition root is not where behaviour is asserted, so nothing covered "an ordinary chat run must not reach the service" as opposed to "produces no outcome object". The cases now drive `__call__` against a recording service, which asserts the same mappings plus what was done with them, and adds the two that were unreachable before: the service left entirely alone for a filtered run, and the completion stamped with a tz-aware current instant. `test_composition.py` gains `TestRunCompletionHook` for the assembly decision that remains -- including that the hook is bound to the service it was given, which a wrong wiring would type-check past. Confirmed by mutation that this case fails when the binding is broken. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> |
||
|
rayhpeng
|
680f044e58 |
refactor(schedule): move schedule_spec translation onto its owners
`spec_column.py` and `spec_wire.py` were the same 60 lines twice, kept in sync by a parity suite. Both are gone; what they did is now split along the line that actually separates the two boundaries. Why two files existed --------------------- They were one module until the slice split it, and the reason given for the split holds: a primary adapter must not import a secondary one, and the two shapes are equal only by coincidence. But that argument only requires the two *shapes* to be independent -- it does not require the parsing *rule* to be written twice, and writing it twice is what needed `test_schedule_spec_parity` to assert the two agreed, down to identical error text. Splitting it properly --------------------- `ScheduleSpec.from_primitives(schedule_type, *, cron, run_at, timezone)` takes four strings, not a `Mapping[str, Any]` -- the mapping was the thing that kept this out of the domain, and four strings carry no transport or storage format with them. It owns the whole rule: unknown type, missing or non-string field, unparseable `run_at`, and (via __post_init__, unchanged) 5-field cron and resolvable timezone. Values are checked rather than trusted, since both callers read data a client can influence. Each adapter keeps only what is genuinely its own -- which two keys its format uses -- as private methods on the class that owns the boundary, matching how `SqlFeedbackRepository` and AWS's own ports-and-adapters sample put the conversion inside the adapter rather than beside it: SqlScheduledTaskRepository._spec_from_row / _spec_to_column models.ScheduledTaskCreateRequest.to_schedule / models._spec_to_wire The emit direction stays duplicated, deliberately: it is three lines per side with no rule in it, and the two are *allowed* to diverge -- one is an HTTP contract, the other a storage format. Asserting they stay byte-identical was a constraint neither side asked for, so that suite is not replaced. The router stops building value objects --------------------------------------- `create` passes `body.to_schedule()`. `update` passes `body.to_schedule(current.schedule)`, replacing eight lines that re-emitted the current spec to the wire shape purely to read defaults back out of it; omitted parts now come off the value object directly. Tests ----- `test_schedule_spec_parity.py` is deleted (162 lines). Its structural and value cases moved to `TestFromPrimitives` in the domain suite -- stated once now instead of parametrized over two implementations. Emit coverage was already elsewhere: wire in `test_schedule_response_models`, column via the repository round-trip in `test_schedule_fakes`. One gap found while removing it: the `once` half of the update fallback had no coverage on either side (both existing cases use cron), and it is exactly the branch this commit rewrites -- `spec_to_wire(current)` round-trip to `current.run_at.isoformat()`. Added `test_a_timezone_change_on_a_once_task_keeps_the_same_instant`, and confirmed by mutation that it is the only case that catches that branch breaking. Behaviour is unchanged: same wire shapes, same normalizations (whitespace in cron, trailing Z re-emitted as +00:00), same error messages, so the 422 details clients see do not move. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> |
||
|
rayhpeng
|
0d86a0d8fe |
fix(schedule): stop the completion hook rolling back the launch write
A cron task could become permanently unschedulable after a run that failed
fast. Reproduced against the real SQL adapters, and against the legacy path
for contrast -- the legacy code does not have this bug, so the hexagonal
slice introduced it.
The window
----------
`dispatch_task` already documents it: "a fast-failing run can reach
handle_run_completion before this write lands". The hook therefore reads the
task while the dispatch path has not yet written its bookkeeping, and the
snapshot it holds still carries the elapsed `next_run_at` the claim was made
on. `handle_run_completion` then wrote that whole snapshot back through
`save()`, so whichever landed second undid the other.
For a cron task the result is terminal in the worst way. `record_launch`
writes the fresh fire time, bumps `run_count` and clears the claim; the
completion's whole-aggregate write restores the elapsed fire time, rolls
`run_count` back and leaves `status='running'` with `lease_expires_at IS
NULL`. Neither `claim_due` branch matches that shape (one needs `enabled`,
the other needs an expired claim), and `cancel_stuck_once_tasks` only sweeps
`once` rows -- so nothing can ever reach the task again.
The legacy `app/scheduler/service.py` wrote the same outcome field by field
(`update(..., updates={"last_error": ...})`) and never touched scheduling
state, which is why it survives the same interleaving.
The fix
-------
`record_completion` joins `record_launch` as a second deliberately narrow
port method, for the same stated reason `record_launch` is not expressed as
`save(task)`: the two race, so neither may write through the aggregate. They
now own disjoint fields -- the launch owns the schedule, the completion owns
the verdict (terminal status plus `last_error`, with `None` meaning "do not
move the status", i.e. every cron task).
`save()` keeps its documented purpose, the user-initiated whole-aggregate
updates (`update_task` / `pause` / `resume`). `handle_run_completion` still
reads the task first, but only to ask `status_after_completion`, which reads
nothing but the schedule type -- immutable, so that read carries no
time-of-check risk.
Tests
-----
The contract suite missed this because it groups cases by port method:
`record_launch` appears only among its own, never interleaved with `save`,
and `test_protect_terminal_keeps_a_concurrently_finalized_verdict` pins the
mirror-image direction only. The in-memory double replaces the whole row too,
so both implementations were faithfully wrong -- the defect was in the
contract, not either adapter.
Added: `TestRecordCompletion` (5 cases, both implementations), including the
interleaving itself; and `test_a_cron_task_survives_a_launch_write_landing_mid_completion`,
which drives a real `ScheduleService` over a repo double that commits
`record_launch` between the hook's read and its write. Both were watched
failing first -- the service case failing with the actual symptom, a rolled
back `next_run_at`, not a missing method.
Verified end to end against file-backed sqlite with the real service: the
task keeps `enabled`, the concurrent fire time and run count survive, the
verdict is recorded, and the next poll still claims it.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
|
||
|
rayhpeng
|
eb37df0779 |
refactor(schedule): wire the slice through a composition root
Switches production onto the hexagonal path. The legacy modules still compile and still have tests, but nothing assembles them any more; deleting them is the next commit, kept separate so it stays reviewable. Composition root ---------------- `app/composition.py::build_domain_services()` is now the only place an adapter is instantiated. It was extracted from `deps.py::langgraph_runtime` rather than added to it: wiring there was tangled with engine startup, orphan recovery and shutdown, so the one rule that governs it -- no SQL backend means no service and the routes answer 503 -- could not be tested without booting the whole application, and was held up by a single comment. It is a pure function of already-built infrastructure, so that rule is now an assertion. Feedback moved with it; doing this while adding schedule's five objects costs one change instead of two. Primary adapter --------------- The router is protocol translation only. What is gone is the giveaway: cron normalisation, `next_run_at` arithmetic, the re-arm rule and hand-written ownership checks all now live in the aggregate. Domain errors map to status codes through one table, so a new error surfaces as a 500 to be classified rather than being swallowed by whichever `except` was nearest. `spec_mapping` split in two (AWS's own layout puts the wire model under the entrypoint that owns it, and a primary adapter must not import a secondary one): `adapters/schedule/spec_column.py` for the JSON column, `routers/schedule/spec_wire.py` for the HTTP body. The two shapes are equal only by coincidence, so `test_schedule_spec_parity.py` runs every case against both and compares their outputs and messages directly. Function names differ per side so an import from the wrong one is visible. Explicit responses ------------------ Routes returned the ORM row's `to_dict()`, leaking `user_id`, `assistant_id`, `overlap_policy` and the two lease columns. The response models publish exactly the field set the frontend declares -- asserted in both directions, since an extra field is a leak and a missing one breaks a client. One wire detail was nearly changed by accident: Pydantic v2 serializes a UTC datetime as `...Z`, while the legacy `coerce_iso` path emitted `+00:00`. `UtcTimestamp` pins `isoformat()` so adopting a model does not silently alter the wire format for every client parsing these. Tests ----- 73 new cases: router behaviour driven through a real `ScheduleService` over in-memory fakes (a mocked service would let the error mapping pass without a domain error ever being raised), response shape, and the composition root. Router mappings verified by mutation -- a wrong status code or a dropped timezone fallback turns 9, 2 and 1 cases red respectively. Two lifespan tests carried a `SimpleNamespace` config that predates this change; `langgraph_runtime` now reads `config.scheduler`, so they were given one. Tolerating the gap with `getattr` was rejected: `AppConfig.scheduler` always exists, so the fallback would be unreachable in production and exist purely to excuse an incomplete test double. Full suite is back to its 24 pre-existing failures. |
||
|
rayhpeng
|
7852421c68 |
feat(schedule): complete the outer ring with the launch adapters
Adds the three remaining adapters plus the poller. Nothing is wired yet -- the composition root is the next commit -- so this is additive and the legacy `app/scheduler/service.py` still serves production. `run_launcher.py` is the pivot of the whole slice. The Gateway signals a busy thread two ways -- `ConflictError` from the run manager, or an `HTTPException(409)` from the route-level path -- which is why the legacy scheduler service imported fastapi to tell them apart. Both are one domain fact, and saying so here is what lets that import disappear without the busy/failed distinction disappearing with it. Everything else becomes `LaunchFailedError`, because the port promises the domain that nothing but its two errors escapes. `CancelledError` is deliberately not caught: shutdown is control flow, not a launch outcome. `thread_lookup.py` narrows `ThreadMetaStore` to the one question this context asks. `require_existing=True` is load-bearing -- the store's default treats an absent row as accessible, which is right for a thread not yet written and wrong for binding a task to it. Both inherit their port explicitly, matching every other adapter in the codebase including feedback's own anti-corruption layer, and both carry the TODO naming the published contract that would replace them once the upstream context has been through a slice of its own. `run_outcome_mapping.py` implements no port: it is the inbound translation the composition root will install on the completion hook, and it owns the filtering the legacy hook did inline. Returning None means "this run is none of the schedule context's business", so the service is simply never called and needs no guard clauses. `poller.py` keeps the two behaviours the legacy loop got right: a failing poll must not end the loop (one transient "database is locked" used to stop scheduling for the rest of the process life), and reconciliation must not block startup. One deliberate behaviour change: the legacy `start()` swept stale runs and stuck once-tasks under separate try/excepts, so the first failing did not stop the second. `reconcile_on_startup` is one call that lets failures propagate -- the domain's position is that fatality is the caller's policy -- so the poller's single except means a failed first sweep now skips the second. Both end up logged and non-fatal, as before. Tests: 50 new cases across the four modules, each port method called and asserted on its return value. That is not decoration: inheriting a Protocol means a misspelled method silently inherits its `...` body and returns None, so the suite was verified by mutation -- renaming `launch` and `exists_for_user` turns 16 and 6 cases red respectively. |
||
|
rayhpeng
|
0bae77ffc0 |
refactor(schedule): move the secondary adapters to app/adapters
Adopts the layout feedback landed in cb49dd67: secondary adapters live under `app/adapters/<context>/`, one file per port, the file named after the port in snake_case with the technology carried by the class name. `app/infra/` is now gone entirely. The rename also separates two meanings of "run" that shared one filename space: `run_sql.py` held `ScheduledRun` (an execution record), while the `run_launcher.py` still to come deals in Gateway runs. task_sql.py -> scheduled_task_repository.py run_sql.py -> scheduled_run_repository.py spec_mapping.py -> spec_mapping.py (implements no port, so no rename) Both SQL adapters now carry the `Secondary adapter (owned persistence)` docstring marker -- this context owns both tables and writes its own queries. `spec_mapping` says instead that it is a boundary mapping and names its two callers. The package `__init__.py` is empty, so imports go through the full path and a class's home file stays unambiguous. Pure move: every top-level symbol was compared against its pre-move original by AST dump (docstrings excluded), plus a separate per-class method-name comparison, since a whole-class dump reports a docstring edit and a renamed method the same way. Also repoints three docstrings and one diagram that still named the deleted `app/infra/` path. |