JjTdZddlmZddlZddlZddlZddlZddlZddlZddl m Z ddl m Z m Z ddlmZmZddlmZddlmZdd lmZdd lmZej2eZe d e efZe eeefge efZd Z Gd deeZ!eGddZ"GddZ#GddeeZ$eGddZ%Gdde#Z&gdZ'y)zLMemoryTaskQueue - in-memory async task queue with lifecycle status tracking.) annotationsN)deque) AwaitableCallable) dataclassfield)Enum)Any)uuid4)SQLiteDatabase.execution_attemptedceZdZdZdZdZdZy) TaskStatuspendingrunning completedfailedN)__name__ __module__ __qualname__PENDINGRUNNING COMPLETEDFAILEDT/home/oem/CyberBrain26-unified/src/cyberbrain26/control_plane/task_queue/__init__.pyrrsGGI FrrceZdZUdZded<ded<ej Zded<dZd ed <dZ d ed <d Z ded<e e jZ ded<dZded<dZded<y) TaskRecordz*A single queued unit of asynchronous work.stridzCallable[[], Awaitable[Any]]funcrstatusNr resultzException | Noneerrorrintattempts)default_factoryfloat created_at float | None started_at completed_at)rrr__doc____annotations__rrr#r$r%r'rtimer*r,r-rrrrr&sf4 G &&#++FJ+FC"E "Hcdii8J8#J #!%L,%rrcneZdZdZd d dZeddZeddZddZddZ ddZ ddZ dd Z dd Z y )MemoryTaskQueueaTAsync FIFO task queue executing callables with status tracking. Tasks are enqueued as callables returning coroutines. `run` drains the pending queue (including tasks enqueued during dispatch), dispatching each through the optional concurrency semaphore and recording PENDING -> RUNNING -> COMPLETED|FAILED transitions. ct|_i|_tjt d||_tj|_d|_ ||_ y)Nr) r_queue _registryasyncio Semaphoremax _semaphoreLock_lock_counter max_retries)selfmax_concurrencyr>s r__init__zMemoryTaskQueue.__init__>sG). 02!++C?,CD\\^  &rc,t|jSN)lenr5r?s rsizezMemoryTaskQueue.sizeFs4;;rcVtd|jjDS)Nc3\K|]$}|jtjk(s!d&yw)r4N)r#rr).0ts r z*MemoryTaskQueue.pending..Ls!XZEWEW9W1Xs",,)sumr6valuesrEs rrzMemoryTaskQueue.pendingJs Xdnn335XXXrcP|xjdz c_tj}t|t j |g|i|}|j j|||j|<tjd||t|j d|S)zFSchedule `func(*args, **kwargs)` for execution and return its task id.r4)r!r"zenqueued task %s)task_id queue_sizeextra) r=r hexr functoolspartialr5appendr6loggerdebugrD)r?r"argskwargsrOrecords renqueuezMemoryTaskQueue.enqueueNs  '++wY->->t-Ud-Uf-UV 6""(w  %S5EF  rcP|jsy|jjS)z@Pop and return the next pending TaskRecord without executing it.N)r5popleftrEs rdequeuezMemoryTaskQueue.dequeue\s{{{{""$$rc8|jj|SrC)r6get)r?rOs rrazMemoryTaskQueue.getbs~~!!'**rcX|jj|}|r |jSdSrC)r6rar#)r?rOrecs rr#zMemoryTaskQueue.statuses'nn  ) szz*d*rc ,Ktj|_tj|_t j d|j|jtjjdd} |j4d{|xjdz c_ |jd{|_ dddd{tj|_tj|_t j d|j|jtjjd|jS777#1d{7swYxYw#t$r}||_tj"|_t j%d|j|dz|j&dz|jtj"jt)|d ||j&k\rtj|_|dz }d d |dz zz}t j+d |j|t-j.|d{7tj|_Yd}~nd}~wwxYww) zExecute a single TaskRecord, updating its status. Returns its result. On failure, retries up to ``self.max_retries`` times with exponential backoff (base delay 0.1s, doubling each attempt). zdispatched task %s)rOr#rQrNr4ztask %s completedztask %s failed (attempt %d/%d))rOr#r%皙?zretrying task %s after %.1fs)rrr#r0r,rWinfor!valuer:r'r"r$rr- Exceptionr%r exceptionr>r rXr7sleep)r?r[attemptexcbackoffs rdispatchzMemoryTaskQueue.dispatchis  #**  IIK II$ii:3E3E3K3KL  ! 3??88OOq(O*0++-$7FM88!+ 4 4 &*iik# 'II&,ii:;O;O;U;UV }}$8$78888 3" * 1 1   4IIaK$$q(#)99","3"3"9"9!$S ! d...*.))+F'1 w{!34 ;VYYPmmG,,, * 2 2 + 3sA9J<E/ E E/)E9E: E E/EBE/JE/EE/E, E# !E,(E// J9C1J *I-+J J JJc Kg}|j4d{|jrR|jj}|j| |j |d{|jrRdddd{|S7t7%#t $rC}t jd|j||jt|dYd}~jd}~wwxYw7Y#1d{7swY|SxYww)ztDrain the queue, dispatching pending tasks (including those enqueued during dispatch) until no tasks remain.Nz$past failure on draining task %s: %s)rOr%rQ) r<r5r^rVrorirWrXr!r )r?executedr[rms rrunzMemoryTaskQueue.runs&(::  ++,,.'--/// ++    0 LL> *0))c#hG !    sC5B C58CB&B 'B+ C8 C5CC5 B C9CCCCC5C2%C( &C2-C5N)r4r)r@r&r>r&returnNonersr&)r"TaskFnrYr rZr rsr )rsTaskRecord | None)rOr rsrw)rOr rszTaskStatus | None)r[rrsr rszlist[TaskRecord])rrrr.rApropertyrFrr\r_rar#rorrrrrr2r25sU'  YY % ++03drr2c$eZdZdZdZdZdZdZdZy)PersistentTaskStatusz6Lifecycle state for JSON-backed jobs stored in SQLite.rrrr interruptedN) rrrr.rrrr INTERRUPTEDrrrr{r{s@GGI FKrr{ceZdZUdZded<ded<ded<ded<d Zd ed <d Zd ed <dZded<dZded<d Z ded<d Z ded<y )PersistentTaskRecordzCSerializable job state that can be inspected after process restart.r r!handlerdict[str, Any]payloadr{r#Nr r$ str | Noner%rr&r'gr)r*r+r,r-) rrrr.r/r$r%r'r*r,r-rrrrrsWM G L   FCE:HcJ#J #!%L,%rrc*eZdZdZdddd dfdZedfd Zedfd Zd d  dd Zdd  dd Z ddZ d dZ d!dd d"dZ d dZ d dZd dZd#dZd$dZd%fd Z d&dZed'dZxZS)(SQLiteTaskQueueaMemory-compatible queue with an additional durable, named-job API. Existing agents may continue to enqueue Python callables in memory. Jobs that must survive restart use :meth:`enqueue_job` with a stable handler key and JSON payload. A process crash changes a running job to ``interrupted``; callers must explicitly retry it to avoid silently repeating side effects. Nr4r)handlersr@r>cFt|||||_t|xsi|_|jj 5}|j dtjjdtjjfdddy#1swYyxYw)N)r@r>z7UPDATE tasks SET status = ?, error = ? WHERE status = ?z?process stopped while task was running; explicit retry required) superrAdatabasedict _handlers transactionexecuter{r}rhr)r?rrr@r> connection __class__s rrAzSQLiteTaskQueue.__init__s kR  8> !'@GH H&tr)rOc  |r|j|k7r tdt|ts t d t j |dd}|xstj}tj}|jj5}|jd|||tjj|fddd|S#ttf$r}t d|d}~wwxYw#1swY|SxYw) z4Persist a named job and a JSON-serializable payload.rz)durable task payload must be a dictionary,:F separators allow_nanz-durable task payload must contain JSON valuesNziINSERT INTO tasks(task_id, handler, payload_json, status, attempts, created_at) VALUES (?, ?, ?, ?, 0, ?))rr isinstancer TypeErrorjsondumpsr rSr0rrrr{rrh) r?rrrO payload_jsonrm identifierr*rs r enqueue_jobzSQLiteTaskQueue.enqueue_jobs'--/W4IJ J'4(GH H V::g*PUVL+ YY[ ]] & & ( J   , (0066   !:& VKLRU U V s#C0C8C5$ C00C58Dc|jj5}|jd|fj}ddd|j |SdS#1swYxYw)Nz%SELECT * FROM tasks WHERE task_id = ?)rrrr_row_to_record)r?rOrrows rget_jobzSQLiteTaskQueue.get_job'sf ]] % % ' :$$7'hj  ,/?t""3'DD   s "AA#c|jj5}|jd|tjj fj }ddd|dk(S#1swYdk(SxYw)aDelete a durable job by task ID. Used by ``cyberbrain plan`` to discard a stale durable result for a planner step before re-planning with the same ``--id``, so the new plan is always freshly dispatched rather than read back from a prior run. Only terminal tasks (completed/failed) are removable; a running task is left untouched so an in-flight dispatch is never orphaned mid-flight. z3DELETE FROM tasks WHERE task_id = ? AND status != ?Nr4)rrrr{rrhrowcount)r?rOrdeleteds r delete_jobzSQLiteTaskQueue.delete_job.sr]] & & ( J ((E(0066 h  !| !|s 7AA,dlimitch|dkr td|jj5}|"|jd|fj }n,|jd|j |fj }dddDcgc]}|j |c}S#1swY(xYwcc}w)Nrzlimit must not be negativez4SELECT * FROM tasks ORDER BY created_at DESC LIMIT ?zESELECT * FROM tasks WHERE status = ? ORDER BY created_at DESC LIMIT ?)rrrrfetchallrhr)r?r#rrrowsrs r list_jobszSQLiteTaskQueue.list_jobsBs 199: : ]] % % ' :~!))JUH(*"))[\\5)(* 599S##C(99  :sAB#B/#B,c|jj5}|jdtjj |tj j fj}ddd|dk(S#1swYdk(SxYw)zrtyperrWrjrkr)r?rrrlr,rclaimedcurrentrhr$rrmr#r-latests rrzSQLiteTaskQueue._dispatch_jobs^ J**,  $,,D-44::",44::  ( !|,,svv.")"5w>3>'  ,**51#$TUU$"jjJRWX H**,  ""7-66<<# ,44::    cff%+v44O  &%)) ]]..0 J&&;1<<BBEFF088>>     1 $"2"22)00-44 (.1E1M1M'MtSWS\S\S^ ]]..0J&&;$s)*<*<)=Ru'E|UXU[U[-55;;=1888$$%=svvFmmC11+=$>???ll366*1c) ,  s2O%AF:.O%>7G 5G6G O%,A$O*O%:G?O%G -O6AI OI O%A6OAM7 OM 9O>O%!O%N(&$O O%OO%O"O%c t|d|dtj|dt|d|dtj|dnd|d|d|d|d |d  S) NrOrrr#rr%r'r*r,r-) r!rrr#r$r%r'r*r,r-)rrrr{)rs rrzSQLiteTaskQueue._row_to_record s~#9~ NJJs>23'H 6585G5S4::c-01Y]g,_<(<(^,  r) rr rz$dict[str, DurableTaskHandler] | Noner@r&r>r&rsrtru)rr rDurableTaskHandlerrboolrsrt)rr rrrOrrsr )rOr rszPersistentTaskRecord | None)rOr rsrrC)r#zPersistentTaskStatus | Nonerr&rslist[PersistentTaskRecord])rsr)rOr rsrrx)rrrrrsr)rr rsr)rrrr.rAryrFrrrrrrrrrrrrrr staticmethodr __classcell__)rs@rrrse:>   7      *3366 ' '$ '  '  '&#     @E,/3: :+: : $ :( *&P 6 O5 !O5$O5  O5b    rr) rrr2rr{rrvrr)(r. __future__rr7rTrrloggingr0 collectionsrcollections.abcrr dataclassesrrenumr typingr uuidr $cyberbrain26.data_plane.sqlite_storer getLoggerrrWrvrr rrrrr2r{rr__all__rrrrsR"  /(?   8 $ #y~% &tCH~. #>? 0d  & &  &wwt 3   & &  &Q oQ h r