iGJdZddlmZddlZddlZddlZddlZddlmZddl m Z m Z ddl m Z ddl mZmZddlmZej$eZe d Zd ZeGd d ZGd dZy)u core/coherence/coherence_orchestrator.py CoherenceOrchestrator — 8-Step Coherence Execution Flow Coordinator. Runs the complete coherence pipeline for a multi-agent swarm round: Step 1 MAP — Push task DAG to Redis Stream via TaskDAGPusher. Step 2 CLAIM — Workers auto-claim tasks via XREADGROUP (background). Step 3 PROPOSE — Workers run and submit StateDelta to StagingArea (background). Step 4 BARRIER — Wait for all worker deltas via StagingArea.wait_for_all. Step 5 REDUCE — OCCCommitEngine collects deltas + SemanticMergeInterceptor merges. Step 6 COMMIT — OCCCommitEngine issues OCC conditional write to RedisMasterState. Step 7 RELEASE — Execute merged side-effects (log to events.jsonl). Step 8 SCAR — On any worker failure, write scar event to events.jsonl. Steps 2 and 3 are implicit — workers are background consumers that run independently once the DAG is pushed to the stream. The orchestrator's role is to coordinate steps 1, 4-8. Total orchestration timeout: 120 seconds (asyncio.wait_for on execute()). Dependency injection: All collaborators are injected via the constructor. None are required at import time — pass mocks in tests. Usage:: orchestrator = CoherenceOrchestrator( dag_pusher=pusher, staging_area=staging, occ_engine=engine, bulkhead=guard, ) result = await orchestrator.execute("sess-abc", tasks=[ {"task_type": "research", "payload": {"query": "..."}}, {"task_type": "synthesize", "payload": {}}, ]) assert result.success # VERIFICATION_STAMP # Story: 6.08 # Verified By: parallel-builder # Verified At: 2026-02-25 # Tests: 22/22 # Coverage: 100% ) annotationsN) dataclass)datetimetimezone)Path)AnyOptional)uuid4z5/mnt/e/genesis-system/data/observability/events.jsonlxcDeZdZUdZded<ded<ded<ded <ded <y ) CoherenceResultaV Result of a CoherenceOrchestrator.execute() call. Attributes: success: True when the full pipeline completed with a committed OCC write. committed_state: The merged patch list from the OCC commit engine. Empty dict when success=False. saga_id: UUID4 identifier for this orchestration run. workers_succeeded: Number of worker tasks (from BulkheadResult) that completed successfully. workers_failed: Number of worker tasks that raised exceptions. boolsuccessdictcommitted_statestrsaga_idintworkers_succeededworkers_failedN)__name__ __module__ __qualname____doc____annotations__>/mnt/e/genesis-system/core/coherence/coherence_orchestrator.pyr r Qs% M Lrr ceZdZdZ d d dZ d dZ d dZ d dZd dZy)CoherenceOrchestratoruL Coordinates the complete 8-step coherence execution pipeline. Constructor args: dag_pusher: TaskDAGPusher — pushes task DAGs to Redis Stream. Must expose: async push_dag(session_id, tasks) -> list[str] staging_area: StagingArea — collects worker deltas in Redis Hash. Must expose: async wait_for_all(session_id, expected_count, timeout_ms) occ_engine: OCCCommitEngine — barrier sync + OCC write coordinator. Must expose: async execute_commit(session_id, expected_workers) -> OccCommitResult bulkhead: BulkheadGuard — asyncio.gather exception isolation. Must expose: async run_with_bulkhead(tasks) -> list[BulkheadResult] qdrant_client: Optional Qdrant client (reserved for future use). NcJ||_||_||_||_||_y)N) dag_pusher staging_area occ_enginebulkhead qdrant_client)selfr"r#r$r%r&s r__init__zCoherenceOrchestrator.__init__}s)%($  *rc Ktt}tjd||t | t j |j|||td{}|S7#t j$r8tjd|t|jd||dtdwxYww)a Run the complete 8-step coherence pipeline with a 120-second timeout. Wraps _execute_pipeline() in asyncio.wait_for so the caller always gets a result within ORCHESTRATION_TIMEOUT_SECONDS. Args: session_id: Session identifier shared by all workers. tasks: List of task dicts, each describing one unit of work. Each task may include keys: task_type, payload, tier, priority. Returns: CoherenceResult describing the final pipeline outcome. Raises: asyncio.TimeoutError: If the pipeline exceeds 120 seconds. The caller should treat this as a failure. zACoherenceOrchestrator: starting saga %s for session %s (%d tasks))timeoutNz2CoherenceOrchestrator: saga %s TIMED OUT after %dsscarorchestration_timeout)r session_idreasontimeout_seconds) rr loggerinfolenasynciowait_for_execute_pipelineORCHESTRATION_TIMEOUT_SECONDS TimeoutErrorerror _write_event)r'r-tasksrresults rexecutezCoherenceOrchestrator.executes.eg, O   J   "++&&z5'B5F, -##  LLD-    &",5'D   ! s/5B>/A0'A.(A0,B>.A00A B;;B>c Ktjd||jE|jj||d{}tjdt ||ntjdtjd|tjd|d}d}|j t |Dcgc]-\}}|jdd ||j|||f/} }}|j j| d{} | D]"\} } t| d s| j$td | D}td | D}tjd |||n t |}d}tjd||jP|jj|t |dd{} tjd|t | ntjdtjd|d}i}|j|jj!|t |d{}|j"r0d|j$i}tj'd||j(n7tj+d||j,ntjdtjd|| |j"nd}|r|j/d|||||dtjd||dkDs|sN||||d}||j"s|j,|d<|j/d |tj+d!||||xr|dk(}t1|||||"}tj'd#|||||S7{cc}}w777fw)$z Internal implementation of the 8-step pipeline. Each step is annotated with its name so the execution order is visible at a glance in logs and in code review. z+CoherenceOrchestrator [Step 1: MAP] saga=%sNz1CoherenceOrchestrator [MAP] pushed %d entries: %szACoherenceOrchestrator [MAP] dag_pusher=None, skipping stream pushuNCoherenceOrchestrator [Step 2: CLAIM] saga=%s (implicit — workers listening)uPCoherenceOrchestrator [Step 3: PROPOSE] saga=%s (implicit — workers proposing)r task_typeztask-closec3:K|]}|jsdywNr.0rs r z:CoherenceOrchestrator._execute_pipeline..s#M!199A#Mc3:K|]}|jrdywrArCrDs rrGz:CoherenceOrchestrator._execute_pipeline..s NqAII NrHzACoherenceOrchestrator [PROPOSE] saga=%s workers: %d ok, %d failedz/CoherenceOrchestrator [Step 4: BARRIER] saga=%si`)expected_count timeout_msz=CoherenceOrchestrator [BARRIER] saga=%s collected %d delta(s)zCCoherenceOrchestrator [BARRIER] staging_area=None, skipping barrierz?CoherenceOrchestrator [Step 5: REDUCE / Step 6: COMMIT] saga=%s)expected_workers merged_patchz>CoherenceOrchestrator [COMMIT] saga=%s committed at version %dz5CoherenceOrchestrator [COMMIT] saga=%s OCC failed: %szCCoherenceOrchestrator [COMMIT] occ_engine=None, skipping OCC commitz/CoherenceOrchestrator [Step 7: RELEASE] saga=%sTrelease)rr-rrrz,CoherenceOrchestrator [Step 8: SCAR] saga=%s)rr-rrocc_failure_reasonr+zLCoherenceOrchestrator [SCAR] saga=%s: %d worker(s) failed, commit_success=%s)rrrrru_CoherenceOrchestrator: saga %s COMPLETE — success=%s, workers_succeeded=%d, workers_failed=%d)r0debugr"push_dagr2r% enumerateget _run_workerrun_with_bulkheadhasattrr?sumr# wait_for_allr$execute_commitrrMr1versionwarning saga_statusr9r )r'r-r:r entry_idsrritask worker_corosbulkhead_results_corodeltas occ_resultrcommit_success scar_payloadoverall_successr;s rr5z'CoherenceOrchestrator._execute_pipelines  BGL ?? &"oo66z5III LLCI  LLS   \    ^   == $ )/  AtHH[E!+6$$Zq9L&*]]%D%D\%RR ( !44)JJL !!$#M/?#M M  N,< NNN LLS!  !$E N  FP    (,,99"5z!:F LLOF   LLU   Mw  " ?? &#==!$U > J!!$2:3J3J"K T&& K** LLU   FP/9/E++4    &",):&4'6    CWM A ^"("0%6 "L %j.@.@5?5K5K 12   fl 3 NN$  )Bn.A #+/)     6       {JV S28s^AON?BO 2O;!OOO6B(OO B O(O)EOO OOc8K|||jddddSw)a Default worker coroutine. In production this would invoke the real Gemini swarm worker logic. In tests, the BulkheadGuard's task list is replaced with mock coroutines. Args: session_id: The orchestration session identifier. task: The task dict from the DAG. task_index: Zero-based index of this task in the DAG. Returns: A dict representing the worker's output (placeholder for now). r>unknown completed)r- task_indexr>status)rS)r'r-r_rls rrTz!CoherenceOrchestrator._run_workers)*%$+y9!   sc tjjdd|tjt j jd|}tjdd5}|jtj|dzd d d tjd |ty #1swY%xYw#t$r!}tjd ||Yd }~y d }~wwxYw) u Append a JSON-lines event to the observability log. The log file is created (including parent directories) on first write. Failures are logged but never propagated — event logging must never crash the orchestration pipeline. Args: event_type: Short label (e.g. "release", "scar"). payload: Arbitrary dict of event metadata. T)parentsexist_ok)tz) event_type timestampazutf-8)encoding Nz+CoherenceOrchestrator: wrote %s event to %sz3CoherenceOrchestrator: failed to write %s event: %s)EVENTS_LOG_PATHparentmkdirrnowrutc isoformatopenwritejsondumpsr0rP Exceptionr8)r'rrpayloadentryfhexcs rr9z"CoherenceOrchestrator._write_events   " " ( ( ( E(%\\X\\:DDFE !%%cG%< 3E*T12 3 LL=  3 3  LLE   s0A/C 1(B=#C =CC C3C..C3)NNNNN) r" Optional[Any]r#rr$rr%rr&rreturnNone)r-rr: list[dict]rr )r-rr:rrrrr )r-rr_rrlrrr)rrrrrrr) rrrrr(r<r5rTr9rrrr r ms "%)&*$("&'+ +! +$ +" +  + % +  +$666  6xOOO O  Oj       >rr )r __future__rr3rloggingos dataclassesrrrpathlibrtypingrr uuidr getLoggerrr0rwr6r r rrrrsw.`#  !'    8 $ NO !$  6qqr