worker.py

Rama main · réplica de lectura del repositorio original.

import asyncio
import logging

from temporalio.worker import Worker

from app.activities.db_activities import (
    acquire_process_run_lease,
    build_process_memory_context,
    compact_episodic_memory,
    confirm_approval_provenance,
    ensure_process_run,
    execute_compensation,
    finalize_task_failure,
    list_process_artifacts,
    promote_completion_memory,
    read_process_memory,
    record_event,
    record_observation,
    record_step,
    register_artifact,
    release_process_run_lease,
    update_process_state,
    update_task_status,
    write_episodic_observation,
    write_process_memory,
)
from app.activities.openclaw_activities import (
    check_intermediate_updates_enabled,
    deliver_reply_files,
    notify_slack_user,
    parse_agent_evaluation,
    send_to_openclaw,
    task_actions_digest,
    validate_openclaw_output,
    verify_response_quality,
)
from app.activities.intake_activities import (
    classify_task_intake_activity,
    resubmit_user_messages,
)
from app.activities.plan_activities import generate_process_plan, save_process_plan
from app.activities.coding_activities import CODING_ACTIVITIES
from app.activities.deep_memory_activities import RECALL_ACTIVITIES
from app.telemetry import init_telemetry
from app.temporal_control import connect_temporal
from app.workflows.catalog_step_child import CatalogStepChildWorkflow
from app.workflows.catalog_task import CatalogTaskWorkflow
from app.workflows.coding_task import CodingTaskWorkflow
from app.workflows.deep_recall import DeepRecallWorkflow
from app.workflows.generic_execute_child import GenericExecuteChildWorkflow
from app.workflows.generic_task import GenericTaskWorkflow
from app.workflows.intake_workflow import IntakeWorkflow

WORKFLOWS = [
    GenericTaskWorkflow,
    CatalogTaskWorkflow,
    CatalogStepChildWorkflow,
    GenericExecuteChildWorkflow,
    IntakeWorkflow,
    DeepRecallWorkflow,
    CodingTaskWorkflow,
]
ACTIVITIES = [
    send_to_openclaw,
    task_actions_digest,
    validate_openclaw_output,
    parse_agent_evaluation,
    update_task_status,
    notify_slack_user,
    deliver_reply_files,
    check_intermediate_updates_enabled,
    verify_response_quality,
    ensure_process_run,
    acquire_process_run_lease,
    release_process_run_lease,
    finalize_task_failure,
    execute_compensation,
    update_process_state,
    record_step,
    record_observation,
    record_event,
    confirm_approval_provenance,
    write_process_memory,
    read_process_memory,
    build_process_memory_context,
    write_episodic_observation,
    compact_episodic_memory,
    promote_completion_memory,
    register_artifact,
    list_process_artifacts,
    generate_process_plan,
    save_process_plan,
    classify_task_intake_activity,
    resubmit_user_messages,
    *RECALL_ACTIVITIES,
    *CODING_ACTIVITIES,
]


async def main():
    logging.info("Starting OpenClaw RMP Worker...")
    init_telemetry("rmp-worker")
    try:
        from app.production.runtime_sync import mark_runtime_boot

        mark_runtime_boot("rmp-worker")
    except Exception as exc:
        logging.warning("runtime boot stamp failed: %s", exc)
    client = await connect_temporal(attempts=8)
    worker = Worker(client, task_queue="openclaw-tasks", workflows=WORKFLOWS, activities=ACTIVITIES)
    from app.deep_memory.curator import warm_up

    warming = asyncio.create_task(warm_up())
    try:
        await worker.run()
    finally:
        warming.cancel()
        from app.decisions.jev import close_jev_client
        from app.llm.openai_direct import close_direct_clients

        await close_jev_client()
        await close_direct_clients()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(main())