$jd(dZddlZddlZddlmZmZddlmZmZddl m Z ej e Z ddededzdefd ZGd d eZGd d eZGddeZGddeZdS)uM Task backend for distributed task processing. This provides an abstraction for task storage and execution: - BrokerTaskBackend: Uses PostgreSQL as broker (production API servers) - WorkerTaskBackend: No-op submit_task (production workers — child tasks are polled) - SyncTaskBackend: Executes tasks immediately (testing/embedded) N)ABCabstractmethod) AwaitableCallable)Anytableschemareturnc&ddlm}|||S)z;Get fully-qualified table name with optional schema prefix.)fq_table_explicit)r r )rr r s i/home/rurouni/.hermes/hermes-agent/venv/lib/python3.11/site-packages/hindsight_api/engine/task_backend.pyfq_tablers'))))))  UF + ++ceZdZdZdZdeeeefge dffdZ e dZ e deeeffdZ e d Zdeeeffd ZdS) TaskBackenda Abstract base class for task execution backends. Implementations must: 1. Store/publish task events (as serializable dicts) 2. Execute tasks through a provided executor callback (optional) The backend treats tasks as pure dictionaries that can be serialized and stored in the database. The executor (typically MemoryEngine.execute_task) receives the dict and routes it to the appropriate handler. c"d|_d|_dS)zInitialize the task backend.NF) _executor _initializedselfs r__init__zTaskBackend.__init__'sMQ!rexecutorNc||_dS)z Set the executor callback for processing tasks. Args: executor: Async function that takes a task dict and executes it N)r)rrs r set_executorzTaskBackend.set_executor,s"rc KdS)zE Initialize the backend (e.g., connect to database). Nrs r initializezTaskBackend.initialize5 r task_dictc KdS)z Submit a task for execution. Args: task_dict: Task as a dictionary (must be serializable) Nrrr s r submit_taskzTaskBackend.submit_task<s  rc KdS)z2 Shutdown the backend gracefully. Nrrs rshutdownzTaskBackend.shutdownFrrcK|j5|dd}td|dS||d{VdS)z Execute a task through the registered executor. Args: task_dict: Task dictionary to execute Raises: Exception: Re-raised from executor on failure. Ntypeunknownz&No executor registered, skipping task )rgetloggerwarningrr task_types r _execute_taskzTaskBackend._execute_taskMsm > !! fi88I NNOIOO P P P FnnY'''''''''''r)__name__ __module__ __qualname____doc__rrdictstrrrrrrr#r%r.rrrrrs  """ "XtCH~.> $.O%P""""  ^  4S>   ^   ^ (T#s(^((((((rrc:eZdZdZdZdeeeffdZdZ dS)SyncTaskBackendz Synchronous task backend that executes tasks immediately. This is useful for tests and embedded/CLI usage where we don't want background workers. Tasks are executed inline rather than being queued. cLKd|_tddS)No-op for sync backend.TzSyncTaskBackend initializedNrr*debugrs rrzSyncTaskBackend.initializegs'  233333rr cK|js|d{V||d{VdS)z Execute the task immediately (synchronously). Args: task_dict: Task dictionary to execute N)rrr.r"s rr#zSyncTaskBackend.submit_tasklsa  $//## # # # # # # #  +++++++++++rcLKd|_tddS)r8FzSyncTaskBackend shutdownNr9rs rr%zSyncTaskBackend.shutdownxs'! /00000rN r/r0r1r2rr3r4rr#r%rrrr6r6_s_444 ,4S> , , , ,11111rr6c:eZdZdZdZdeeeffdZdZ dS)WorkerTaskBackendu; Task backend for worker processes. Workers execute tasks directly via the poller (claim → execute), so they don't need submit_task to run anything. When engine code running *inside* a worker-executed task calls submit_task (e.g. retain triggers consolidation), the async-operation row has already been persisted (with task_payload) by _submit_async_operation — so submit_task is a no-op. The new task will be picked up by a worker on the next poll cycle instead of being executed inline, which avoids blocking the parent task. cLKd|_tddS)NTzWorkerTaskBackend initializedr9rs rrzWorkerTaskBackend.initializes'  455555rr crK|dd}td|ddS)zJNo-op: the row already exists in async_operations; a worker will claim it.r'r(z)WorkerTaskBackend: submit_task no-op for z (will be picked up by poller)N)r)r*r:r,s rr#zWorkerTaskBackend.submit_tasks=MM&)44  jjjjkkkkkrcLKd|_tddS)NFzWorkerTaskBackend shutdownr9rs rr%zWorkerTaskBackend.shutdowns'! 122222rNr=rrrr?r?~sd  666l4S>llll 33333rr?c eZdZdZ ddegefdedzdegedzfdzffd ZdZde eeffd Z d Z dd e fd Z xZS)BrokerTaskBackenda6 Task backend using PostgreSQL as broker. submit_task() stores task_payload in async_operations table. Actual polling and execution is handled separately by WorkerPoller. This backend is used by the API to store tasks. Workers poll the database separately to claim and execute tasks. N pool_getterr schema_gettercrt||_||_||_dS)a Initialize the broker task backend. Args: pool_getter: Callable that returns the asyncpg connection pool schema: Database schema for multi-tenant support (optional, static) schema_getter: Callable that returns current schema dynamically (optional). If set, takes precedence over static schema for submit_task. N)superr _pool_getter_schema_schema_getter)rrEr rF __class__s rrzBrokerTaskBackend.__init__s8 ' +rcLKd|_tddS)zInitialize the backend.TzBrokerTaskBackend initializedNrr*infors rrzBrokerTaskBackend.initializes'  344444rr cK|js|d{V|}|d}|dd}|d}ddlmfd}t j|| }|jr|n|j}td |} d d l m } |rv| |4d{V} | d | d||d{Vdddd{Vn#1d{VswxYwYtd|ddSddl} | } | |4d{V} | d| d| |||d{Vdddd{Vn#1d{VswxYwYtd| d|dS)a/ Store task payload in async_operations table. The task_dict should contain an 'operation_id' if updating an existing operation record, otherwise a new operation will be created. Args: task_dict: Task dictionary to store (must be JSON serializable) N operation_idr'r(bank_idr)datetimect|r|Stdt|jd)NzObject of type z is not JSON serializable) isinstance isoformat TypeErrorr'r/)objrSs rdatetime_encoderz7BrokerTaskBackend.submit_task..datetime_encodersF#x(( '}}&[d3ii.@[[[\\ \r)defaultasync_operationsr acquire_with_retryz UPDATE z SET task_payload = $1::jsonb, updated_at = now() WHERE operation_id = $2 AND task_payload IS NULL z!submit_task UPDATE for operation z (no-op if payload already set)z! INSERT INTO z (operation_id, bank_id, operation_type, status, task_payload) VALUES ($1, $2, $3, 'pending', $4::jsonb) zCreated new operation z for task type )rrrIr)rSjsondumpsrKrJrdb_utilsr]executer*r:uuiduuid4)rr poolrQr-rRrY payload_jsonr rr]connrbnew_idrSs @rr#zBrokerTaskBackend.submit_tasks  $//## # # # # # # #  "" }}^44 MM&)44 -- ** &%%%%% ] ] ] ] ] z)5EFFF *.*=O$$&&&4<+V44000000 " V *)$//       4ll! !                             LLj\jjj k k k k k KKKZZ\\F))$//       4ll!&                                      LLT&TTTT U U U U Us$%"D D#&D#3$F)) F36F3cLKd|_tddS)zShutdown the backend.FzBrokerTaskBackend shutdownNrNrs rr%zBrokerTaskBackend.shutdowns'! 011111r^@timeoutcKddl}ddlm}|}|jr|n|j}t d|}|}||z |kr||4d{V}| d|dd{V} dddd{Vn#1d{VswxYwY| dkrdS| dd{V||z |kt d |d dS) aK Wait for pending tasks to be processed. In the broker model, this polls the database to check if tasks for this process have been completed. This is useful in tests when worker_enabled=True (API processes its own tasks). Args: timeout: Maximum time to wait in seconds rNr r\r[z* SELECT COUNT(*) FROM z_ WHERE status = 'pending' AND task_payload IS NOT NULL g?z(Timeout waiting for pending tasks after s) asyncior`r]rIrKrJrget_event_looptimefetchvalsleepr*r+) rrjrmr]rdr r start_timerfcounts rwait_for_pending_tasksz(BrokerTaskBackend.wait_for_pending_taskss2 000000  ""*.*=O$$&&&4<+V44++--2244 $$&&++-- :WDD))$//       4"mm*/                           zz--$$ $ $ $ $ $ $ $$$&&++-- :WDD L'LLLMMMMMs7 C)) C36C3)NN)ri)r/r0r1r2rrr4rrr3r#r%floatrt __classcell__)rLs@rrDrDs"9= ,,b#g&,d , C$J/$6 ,,,,,,(555 CV4S>CVCVCVCVJ222 #N#NE#N#N#N#N#N#N#N#NrrD)N)r2r^loggingabcrrcollections.abcrrtypingr getLoggerr/r*r4rrr6r?rDrrrr|sx ########////////  8 $ $,,C,t,s,,,,B(B(B(B(B(#B(B(B(J11111k111>33333 3336QNQNQNQNQN QNQNQNQNQNr