Skip to content

Commit 9df2f33

Browse files
DeepfreezechillBrian KrafftCopilot
authored
feat(5.2): Extract EvolutionOrchestrator into evolution/orchestrator.py (#40)
* feat(5.2): Extract EvolutionOrchestrator into evolution/orchestrator.py - New: evolution/orchestrator.py (130 lines) — dispatch_evolution, execute_contexts, schedule_background, log_background_result - Modified: evolver.py — evolve(), _execute_contexts(), schedule_background(), _log_background_result delegate to orchestrator - New: tests/test_evolution_orchestrator.py — 18 tests (dispatch routing, throttling, background tasks, backward compat) - Suite: 1,443 passed, 127 skipped Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * fix(5.2): Address review findings — preserve evolve() call chain, fix logger, add delegation tests - F1 HIGH: execute_contexts now calls evolver.evolve() not dispatch_evolution() directly - F2 MEDIUM: Logger changed to Logger.get_logger(__name__) (repo standard) - F5 LOW: Added 3 TestDelegationSeam tests (real SkillEvolver → orchestrator path) - Suite: 1,446 passed, 127 skipped Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --------- Co-authored-by: Brian Krafft <bkrafft@microsoft.com> Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
1 parent 907e2ae commit 9df2f33

3 files changed

Lines changed: 608 additions & 58 deletions

File tree

‎openspace/skill_engine/evolution/orchestrator.py‎

Lines changed: 148 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,148 @@
1+
"""Evolution orchestration — dispatch, throttle, and background scheduling.
2+
3+
Functions extracted from ``SkillEvolver`` (Epic 5.2):
4+
- ``dispatch_evolution`` — route a single context to the right engine
5+
- ``execute_contexts`` — run a batch of contexts in parallel (throttled)
6+
- ``schedule_background`` — fire-and-forget background evolution task
7+
- ``log_background_result`` — callback that logs task outcome
8+
"""
9+
10+
from __future__ import annotations
11+
12+
import asyncio
13+
import logging
14+
from typing import TYPE_CHECKING, List, Optional
15+
16+
from openspace.utils.logging import Logger
17+
18+
from .models import EvolutionContext
19+
20+
from openspace.skill_engine.types import EvolutionType
21+
22+
if TYPE_CHECKING:
23+
from openspace.skill_engine.types import SkillRecord
24+
25+
logger = Logger.get_logger(__name__)
26+
27+
28+
# ---------------------------------------------------------------------------
29+
# Single-context dispatch
30+
# ---------------------------------------------------------------------------
31+
32+
async def dispatch_evolution(
33+
evolver,
34+
ctx: EvolutionContext,
35+
) -> Optional["SkillRecord"]:
36+
"""Execute one evolution action. Returns new ``SkillRecord`` or *None*.
37+
38+
Delegates to the appropriate engine method on *evolver*:
39+
``_evolve_fix``, ``_evolve_derived``, or ``_evolve_captured``.
40+
41+
The global semaphore is **not** acquired here — it is managed by
42+
``execute_contexts`` so the concurrency limit covers whole batches.
43+
"""
44+
evo_type: EvolutionType = ctx.suggestion.evolution_type
45+
try:
46+
if evo_type == EvolutionType.FIX:
47+
return await evolver._evolve_fix(ctx)
48+
elif evo_type == EvolutionType.DERIVED:
49+
return await evolver._evolve_derived(ctx)
50+
elif evo_type == EvolutionType.CAPTURED:
51+
return await evolver._evolve_captured(ctx)
52+
else:
53+
logger.warning("Unknown evolution type: %s", evo_type)
54+
return None
55+
except Exception as e:
56+
targets = "+".join(ctx.suggestion.target_skill_ids) or "(new)"
57+
logger.error(
58+
"Evolution failed [%s] target=%s: %s",
59+
evo_type.value,
60+
targets,
61+
e,
62+
)
63+
return None
64+
65+
66+
# ---------------------------------------------------------------------------
67+
# Batch execution with throttling
68+
# ---------------------------------------------------------------------------
69+
70+
async def execute_contexts(
71+
evolver,
72+
contexts: List[EvolutionContext],
73+
trigger_label: str,
74+
) -> List["SkillRecord"]:
75+
"""Execute a list of evolution contexts in parallel (throttled).
76+
77+
Uses *evolver._semaphore* for concurrency control and routes each
78+
context through ``dispatch_evolution``.
79+
"""
80+
81+
async def _throttled(c: EvolutionContext) -> Optional["SkillRecord"]:
82+
async with evolver._semaphore:
83+
return await evolver.evolve(c)
84+
85+
raw = await asyncio.gather(
86+
*[_throttled(c) for c in contexts],
87+
return_exceptions=True,
88+
)
89+
results: List["SkillRecord"] = []
90+
for r in raw:
91+
if isinstance(r, BaseException):
92+
logger.error("[Trigger:%s] Evolution task raised: %s", trigger_label, r)
93+
elif r is not None:
94+
results.append(r)
95+
96+
if results:
97+
names = [r.name for r in results]
98+
logger.info(
99+
"[Trigger:%s] Evolved %d skill(s): %s",
100+
trigger_label,
101+
len(results),
102+
names,
103+
)
104+
return results
105+
106+
107+
# ---------------------------------------------------------------------------
108+
# Background task scheduling
109+
# ---------------------------------------------------------------------------
110+
111+
def schedule_background(
112+
background_tasks: set,
113+
coro,
114+
*,
115+
label: str = "background_evolution",
116+
) -> Optional[asyncio.Task]:
117+
"""Launch *coro* as a background ``asyncio.Task``.
118+
119+
*background_tasks* is the ``set`` that tracks outstanding tasks
120+
(typically ``SkillEvolver._background_tasks``). The task is removed
121+
from the set on completion via ``discard`` callback.
122+
"""
123+
try:
124+
loop = asyncio.get_running_loop()
125+
except RuntimeError:
126+
logger.warning("No running event loop — cannot schedule %s", label)
127+
return None
128+
129+
task = loop.create_task(coro, name=label)
130+
background_tasks.add(task)
131+
task.add_done_callback(background_tasks.discard)
132+
task.add_done_callback(log_background_result)
133+
return task
134+
135+
136+
def log_background_result(task: asyncio.Task) -> None:
137+
"""Log the outcome of a background evolution task."""
138+
if task.cancelled():
139+
logger.debug("Background task '%s' was cancelled", task.get_name())
140+
return
141+
exc = task.exception()
142+
if exc:
143+
logger.error(
144+
"Background task '%s' failed: %s",
145+
task.get_name(),
146+
exc,
147+
exc_info=exc,
148+
)

‎openspace/skill_engine/evolver.py‎

Lines changed: 12 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,12 @@
3333
EvolutionTrigger,
3434
_sanitize_skill_name,
3535
)
36+
from .evolution.orchestrator import (
37+
dispatch_evolution as _dispatch_evolution,
38+
execute_contexts as _execute_contexts_impl,
39+
log_background_result as _log_background_result_impl,
40+
schedule_background as _schedule_background_impl,
41+
)
3642
from .patch import (
3743
SKILL_FILENAME,
3844
PatchType,
@@ -181,21 +187,7 @@ async def evolve(self, ctx: EvolutionContext) -> Optional[SkillRecord]:
181187
The global semaphore is NOT acquired here — it is managed at the
182188
trigger-method level so the concurrency limit covers the whole batch.
183189
"""
184-
evo_type = ctx.suggestion.evolution_type
185-
try:
186-
if evo_type == EvolutionType.FIX:
187-
return await self._evolve_fix(ctx)
188-
elif evo_type == EvolutionType.DERIVED:
189-
return await self._evolve_derived(ctx)
190-
elif evo_type == EvolutionType.CAPTURED:
191-
return await self._evolve_captured(ctx)
192-
else:
193-
logger.warning(f"Unknown evolution type: {evo_type}")
194-
return None
195-
except Exception as e:
196-
targets = "+".join(ctx.suggestion.target_skill_ids) or "(new)"
197-
logger.error(f"Evolution failed [{evo_type.value}] target={targets}: {e}")
198-
return None
190+
return await _dispatch_evolution(self, ctx)
199191

200192
# Trigger 1: post-analysis
201193
async def process_analysis(
@@ -441,26 +433,7 @@ async def _execute_contexts(
441433
442434
Used by all three triggers after building/confirming contexts.
443435
"""
444-
445-
async def _throttled(c: EvolutionContext) -> Optional[SkillRecord]:
446-
async with self._semaphore:
447-
return await self.evolve(c)
448-
449-
raw = await asyncio.gather(
450-
*[_throttled(c) for c in contexts],
451-
return_exceptions=True,
452-
)
453-
results: List[SkillRecord] = []
454-
for r in raw:
455-
if isinstance(r, BaseException):
456-
logger.error(f"[Trigger:{trigger_label}] Evolution task raised: {r}")
457-
elif r is not None:
458-
results.append(r)
459-
460-
if results:
461-
names = [r.name for r in results]
462-
logger.info(f"[Trigger:{trigger_label}] Evolved {len(results)} skill(s): {names}")
463-
return results
436+
return await _execute_contexts_impl(self, contexts, trigger_label)
464437

465438
def schedule_background(
466439
self,
@@ -474,30 +447,11 @@ def schedule_background(
474447
``background_triggers`` is True. The task is tracked so it can
475448
be awaited on shutdown via ``wait_background()``.
476449
"""
477-
try:
478-
loop = asyncio.get_running_loop()
479-
except RuntimeError:
480-
logger.warning(f"No running event loop — cannot schedule {label}")
481-
return None
482-
483-
task = loop.create_task(coro, name=label)
484-
self._background_tasks.add(task)
485-
task.add_done_callback(self._background_tasks.discard)
486-
task.add_done_callback(self._log_background_result)
487-
return task
450+
return _schedule_background_impl(
451+
self._background_tasks, coro, label=label,
452+
)
488453

489-
@staticmethod
490-
def _log_background_result(task: asyncio.Task) -> None:
491-
"""Log the outcome of a background evolution task."""
492-
if task.cancelled():
493-
logger.debug(f"Background task '{task.get_name()}' was cancelled")
494-
return
495-
exc = task.exception()
496-
if exc:
497-
logger.error(
498-
f"Background task '{task.get_name()}' failed: {exc}",
499-
exc_info=exc,
500-
)
454+
_log_background_result = staticmethod(_log_background_result_impl)
501455

502456
# LLM confirmation for Trigger 2/3
503457
async def _llm_confirm_evolution(

0 commit comments

Comments
 (0)