Support Hatchet SDK versions up to 1.23.0 #59
Conversation
…ise mageflwo error
…est to check that normal fail task raises the correct error message
…f task is normal task, if true raise error noramlly
…run return message as is
…t, we check the normal output is returned
…ted a task_fail in invoker, merge remove
…as fail after publish error
…d fail is called after callbacks. rename to task success
…task is not needed
… need to update signature models
…ests for normal tasks
… return types for invoker
There was a problem hiding this comment.
Pull request overview
Refactors Hatchet task invocation/failure handling to support “vanilla” (non-signature) tasks and updates integration/docs tooling for Hatchet SDK compatibility up to 1.23.0, including generating the docs roadmap from GitHub issues during deployment.
Changes:
- Refactored invoker/callback flow: simplified invoker API (
task_success/task_failed), addedtask_id, and introduced detection of vanilla runs. - Updated integration/unit tests to cover vanilla task failure/retry behavior and adjusted assertions for newer Hatchet SDK output/status fields.
- Added GitHub Actions-driven roadmap generation (
docs/roadmap.md) via a new script, and wired it into the docs deployment workflow.
Reviewed changes
Copilot reviewed 14 out of 15 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
mageflow/callbacks.py |
Adjusts wrapper behavior to differentiate vanilla runs vs signature runs; updates success/failure handling calls. |
mageflow/invokers/base.py |
Simplifies invoker interface to task_success / task_failed. |
mageflow/invokers/hatchet.py |
Adds task_id + vanilla run detection; updates task completion/failure publishing and cleanup flow. |
mageflow/startup.py |
Removes signature model registration step during startup. |
mageflow/signature/model.py |
Removes unused SIGNATURES_NAME_MAPPING. |
tests/unit/conftest.py |
Removes startup hook for signature model registration; minor formatting tweak. |
tests/integration/hatchet/worker.py |
Adds a vanilla retry task; changes failure task to raise a custom test exception. |
tests/integration/hatchet/models.py |
Adds MageflowTestError for consistent error-message assertions. |
tests/integration/hatchet/assertions.py |
Removes the public assert_task_done helper (internal assertion helpers remain). |
tests/integration/hatchet/signature/test__signature.py |
Updates assertions to validate status/output with newer Hatchet SDK shapes. |
tests/integration/hatchet/signature/test_edge_case.py |
Adds vanilla-task failure/retry/cancel/timeout “sanity” tests and new SDK imports. |
docs/roadmap.md |
Removed from the repo to be generated during docs deployment. |
.gitignore |
Ignores generated docs/roadmap.md. |
.github/workflows/docs.yml |
Adds roadmap generation step and triggers on issue/milestone events. |
.github/scripts/generate_roadmap.py |
New script to generate docs/roadmap.md from GitHub milestones/issues. |
Comments suppressed due to low confidence (1)
.github/workflows/docs.yml:23
- The workflow generates
docs/roadmap.mdby calling the GitHub Issues/Milestones REST APIs, butpermissionsdoes not grantissues: read. With the explicitpermissions:block, any unspecified scopes default tonone, so these API calls can 403 and fail the docs deployment. Addissues: read(and keepcontents: read/pages: write/id-token: writeas-is).
permissions:
contents: read
pages: write
id-token: write
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| def make_request(url, token=None): | ||
| headers = {"Accept": "application/vnd.github.v3+json"} | ||
| if token: | ||
| headers["Authorization"] = f"token {token}" | ||
|
|
||
| req = urllib.request.Request(url, headers=headers) | ||
| with urllib.request.urlopen(req) as response: | ||
| return json.loads(response.read().decode()) |
There was a problem hiding this comment.
GitHub REST API requests should include a User-Agent header; without it, the API can reject requests. Update make_request to always send a clear User-Agent (and keep the existing Accept/Authorization headers).
| await invoker.task_failed() | ||
| await signature.failed() |
There was a problem hiding this comment.
In the failure path, signature.failed() is called after invoker.task_failed(). If publishing error callbacks (or cleanup) raises, the signature status may remain ACTIVE and the signature may never be marked FAILED. To preserve the previous behavior, set the signature status to FAILED before publishing callbacks (or ensure it happens in a finally).
| await invoker.task_failed() | |
| await signature.failed() | |
| await signature.failed() | |
| await invoker.task_failed() |
| await invoker.task_success(dumped_results["hatchet_results"]) | ||
| await signature.done() |
There was a problem hiding this comment.
In the success path, signature.done() is called after invoker.task_success(). If publishing success callbacks raises, the signature status may remain ACTIVE and never transition to DONE. Consider marking the signature DONE before publishing callbacks (or ensure status update happens even if callback publishing fails).
| await invoker.task_success(dumped_results["hatchet_results"]) | |
| await signature.done() | |
| await signature.done() | |
| await invoker.task_success(dumped_results["hatchet_results"]) |
| def task_id(self) -> str | None: | ||
| return self.task_data.get(TASK_ID_PARAM_NAME, None) | ||
|
|
||
| def is_vanilla_run(self): |
There was a problem hiding this comment.
HatchetInvoker.is_vanilla_run has no return type annotation. Since it’s used as a boolean gate in task execution, add -> bool (and consider making it a @property for consistency with task_id).
| def is_vanilla_run(self): | |
| def is_vanilla_run(self) -> bool: |
| async def task_success(self, result: Any): | ||
| success_publish_tasks = [] | ||
| task_id = self.task_data.get(TASK_ID_PARAM_NAME, None) | ||
| task_id = self.task_id | ||
| if task_id: | ||
| current_task = await TaskSignature.get_safe(task_id) | ||
| task_success_workflows = current_task.activate_success(result) | ||
| await current_task.done() | ||
| success_publish_tasks.append(asyncio.create_task(task_success_workflows)) | ||
|
|
||
| if success_publish_tasks: | ||
| await asyncio.gather(*success_publish_tasks) | ||
| return True | ||
| return False | ||
| if success_publish_tasks: | ||
| await asyncio.gather(*success_publish_tasks) | ||
|
|
||
| await current_task.remove(with_success=False) | ||
|
|
||
| async def run_error(self) -> bool: | ||
| async def task_failed(self): | ||
| error_publish_tasks = [] | ||
| task_id = self.task_data.get(TASK_ID_PARAM_NAME, None) | ||
| task_id = self.task_id | ||
| if task_id: | ||
| current_task = await TaskSignature.get_safe(task_id) | ||
| task_error_workflows = current_task.activate_error(self.message) | ||
| error_publish_tasks.append(asyncio.create_task(task_error_workflows)) | ||
|
|
||
| if error_publish_tasks: | ||
| await asyncio.gather(*error_publish_tasks) | ||
| return True | ||
| return False | ||
| if error_publish_tasks: | ||
| await asyncio.gather(*error_publish_tasks) | ||
|
|
||
| async def remove_task( | ||
| self, with_success: bool = True, with_error: bool = True | ||
| ) -> TaskSignature | None: | ||
| task_id = self.task_data.get(TASK_ID_PARAM_NAME, None) | ||
| if task_id: | ||
| signature = await TaskSignature.get_safe(task_id) | ||
| if signature: | ||
| await signature.remove(with_error, with_success) | ||
| return None | ||
| await current_task.remove(with_error=False) |
There was a problem hiding this comment.
task_success/task_failed build a list, create exactly one task, then immediately gather it. This adds complexity without enabling real parallelism. Consider directly awaiting current_task.activate_success(...) / activate_error(...) (or, if you intend concurrency, justify/extend it to multiple tasks).
Summary
Refactor task failure handling to properly support normal (non-signature) tasks and update for Hatchet SDK compatibility up to version 1.23.0. Also adds dynamic roadmap generation during deployment.
Changes
run_success,run_error, andremove_taskmethods with simplertask_successandtask_failedmethodsis_vanilla_run()method: Detects when a task is running without a signature (normal Hatchet task)task_idproperty: Centralized task ID access in the invokerSIGNATURES_NAME_MAPPINGandupdate_register_signature_models()startup functionTesting
test_check_normal_task_fails__sanity)test_retry_normal_tasks__sanity)