feat(workflows): TaskExecutorWorkflow stub + wire worker — T19/T20

activities.py — persist_task_instance (new):
  Idempotent INSERT ... ON CONFLICT (id) DO NOTHING on task_instances.
  task_id passed in from workflow (derived from workflow_id via uuid5).
  Registered on task-execution-tq.

workflows.py — TaskExecutorWorkflow (T19):
  Derives stable task_id = uuid5(NAMESPACE_URL, workflow_id).
  Calls persist_task_instance → status=done, returns immediately.
  Real execution logic to replace stub in a later workstream.

worker.py — T20:
  Registers persist_task_instance on task-execution-tq Worker.
  Both queues fully wired: orchestrator-tq and task-execution-tq.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-26 22:30:50 +00:00
parent da7de6ea3b
commit 34aa70cbd9
4 changed files with 75 additions and 6 deletions

View File

@@ -21,6 +21,7 @@ with workflow.unsafe.imports_passed_through():
from activity_core.activities import (
load_activity_definition,
log_run,
persist_task_instance,
resolve_context,
)
from activity_core.template_engine import evaluate_templates
@@ -123,9 +124,38 @@ class RunActivityWorkflow:
class TaskExecutorWorkflow:
"""Child workflow that executes one concrete task instance.
Stub implementation — T19.
Stub behaviour: persists a task_instances row with status=done and
returns immediately. Real task execution logic replaces this in a
later workstream.
task_id is derived deterministically from the workflow's own ID so
persist_task_instance retries remain idempotent.
"""
@workflow.run
async def run(self, run_id: str, task_type: str, params: dict) -> dict:
raise NotImplementedError("T19")
# Derive a stable task_id from this workflow's own ID.
task_id = str(
uuid.uuid5(uuid.NAMESPACE_URL, workflow.info().workflow_id)
)
workflow.logger.info(
"TaskExecutorWorkflow started",
extra={"run_id": run_id, "task_type": task_type, "task_id": task_id},
)
await workflow.execute_activity(
persist_task_instance,
{
"id": task_id,
"run_id": run_id,
"type": task_type,
"params": params,
"status": "done",
},
task_queue=_TASK_QUEUE,
start_to_close_timeout=_ACTIVITY_TIMEOUT,
retry_policy=_RETRY_POLICY,
)
return {"task_id": task_id, "status": "done"}