Seed(무엇을 만들지 적은 명세서) 안의 합격 기준들을 의존성에 따라 레벨로 나눠 AI 에이전트를 병렬로 돌리는 실행 엔진이다.
왜 배우나: “한 번의 명령으로 여러 작업을 동시에·안전하게 돌려 코드를 완성하는” ouroboros의 심장부이기 때문이다.
그림
flowchart TD
A["ooo run → execute_seed 호출"] --> B["prepare_session: 세션·취소·하트비트 등록"]
B --> C["프롬프트 만들기<br/>build_system_prompt / build_task_prompt"]
C --> D{"합격기준 2개 이상<br/>or fat-harness 모드?"}
D -- 예 --> E["DependencyAnalyzer.analyze<br/>LLM에게 의존성 질의 → 의존성 그래프"]
E --> F["to_execution_plan<br/>레벨 그래프 만들기"]
F --> G[레벨을 위에서 아래로 직렬 루프]
G --> H["한 레벨 안의 합격기준들은 병렬 실행<br/>_execute_atomic_ac → adapter.execute_task"]
H --> I["메시지마다 stall 타이머 리셋<br/>+ 하트비트 + 진행 이벤트"]
I --> J["코디네이터: 파일충돌 감지·중재 리뷰 게이트"]
J --> K{"다음 레벨 있나?"}
K -- 예 --> G
K -- 아니오 --> L[OrchestratorResult 반환]
M["Watchdog 벽시계 / 취소 레지스트리"] -. 항상 감시 .- G
쉽게 풀기
흔한 오해부터 깨자. “AI에게 일 시키기 = 프롬프트 한 번 던지고 답 한 번 받기”가 아니다. 요리 주방에 비유하면 쉽다.
주문서(Seed)를 받는다. “무엇을 만들지(goal)“와 “완성 조건들(합격 기준, Acceptance Criteria=AC, 보통 여러 개)“이 들어 있다.
요리 순서를 파악한다(의존성 분석). “소스를 졸이려면 먼저 육수를 내야 한다”처럼 선후 관계를 LLM에게 직접 물어 의존성 그래프를 만든다.
동시에 할 것끼리 묶는다(레벨/스테이지). 서로 안 막는 작업은 한 레벨로. 야채 썰기와 물 끓이기는 같은 레벨이다. 이 레벨들이 위→아래로 줄선다.
레벨 하나를 통째로 병렬 조리한다. 한 레벨 안 AC들은 각자 별도 에이전트(요리사)가 동시에 작업하고, 각자는 주문서 전체가 아니라 자기 AC 한 개 + 옆 요리사 경계 + 직전 단계 정보만 받는다 → 충돌·혼선 감소.
레벨이 끝나면 심판(코디네이터)이 검사한다. “두 요리사가 같은 도마(파일)를 동시에 썼나?”를 보고, 충돌이 있으면 중재 AI 세션으로 정리·경고를 다음 레벨에 전달한다. 충돌 없으면 이 비싼 단계는 건너뛴다(비용 0).
성적표를 낸다. 전부 성공했는지 판정해 OrchestratorResult(성공·세션ID·요약·소요시간)로 반환한다.
전 과정을 OrchestratorRunner.execute_seed()가 지휘하고, 도는 내내 watchdog·하트비트·취소 레지스트리가 멈춤/타임아웃/취소를 따로 감시한다.
flowchart LR
subgraph L0["레벨0 · 병렬"]
A0[AC1]
A1[AC2]
end
subgraph L1["레벨1 · 병렬"]
A2[AC3]
end
L0 -->|코디네이터 게이트| L1
A0 -. 자기 AC+경계+직전컨텍스트만 .-> A0
핵심 모양: "단일 호출"이 아니라 레벨 그래프를 위→아래 한 줄씩, 각 줄은 병렬로 도는 루프다. 세로는 직렬, 가로는 병렬.
ACNode/ExecutionStage/StagedExecutionPlan (dependency_analyzer.py:74,118,137) — AC-tree 레벨 그래프
이름
타입
설명
ACNode.index / content
int / str
AC 번호(0-based) / 본문
ACNode.depends_on
tuple[int,…]
선행 AC 인덱스들
ACNode.can_run_independently
bool
병렬 스테이지에 넣어도 되는지
ExecutionStage.ac_indices
tuple[int,…]
한 레벨에서 동시에 도는 AC들
ExecutionStage.depends_on_stages
tuple[int,…]
선행 스테이지
StagedExecutionPlan.stages
tuple[…]
직렬 스테이지 목록 = 레벨 그래프
execution_levels (prop)
tuple[…]
레거시 레벨 뷰
FileConflict/CoordinatorReview (coordinator.py:85,102) — 레벨 사이 리뷰 게이트
이름
타입
설명
FileConflict.file_path
str
충돌 파일 경로
FileConflict.ac_indices
tuple[int,…]
그 파일을 건드린 AC들(2개↑=충돌)
FileConflict.resolved
bool
코디네이터 해결 여부
CoordinatorReview.warnings_for_next_level
tuple[str,…]
다음 레벨에 주입될 경고
CoordinatorReview.fixes_applied
tuple[str,…]
적용된 수정 설명
생명주기(트리거→반환)는 한눈에 보자.
sequenceDiagram
participant U as 사용자(ooo run)
participant R as Runner
participant A as DependencyAnalyzer
participant P as ParallelACExecutor
participant C as Coordinator
U->>R: execute_seed(seed, exec_id)
R->>R: prepare_session(세션·취소·하트비트)
R->>R: build_system/task_prompt
R->>A: AC 2개↑ or fat-harness → analyze
A-->>R: 레벨 그래프(StagedExecutionPlan)
loop 스테이지 직렬
R->>P: 레벨 AC 병렬 실행
P->>C: 레벨 끝 → 충돌 감지·중재
end
R-->>U: OrchestratorResult
펼쳐보기: 단계별 상세 (함수·폴백 규칙)
트리거 — ooo run(또는 ooo auto 실행 단계) → MCP 핸들러가 execute_seed(seed, execution_id) 호출. 생명주기 시작점.
세션 준비 — prepare_session이 이벤트 스토어에 세션(exec_<hex12>) 생성, 취소·하트비트 추적 등록.
Seed→프롬프트 — execute_precreated_session이 seed.task_type으로 전략 선택 → build_system_prompt(전략 fragment+SeedContract+AC 추적+복구 프로토콜), build_task_prompt(Goal+번호 AC).
분기 — AC 2개↑ 또는 fat-harness면 _execute_parallel로.
레벨 그래프 — DependencyAnalyzer.analyze가 LLM에 의존성을 JSON으로 질의 → to_execution_plan(). 분석 실패 시 전체 AC를 단일 병렬 레벨로 폴백, --sequential이면 각 AC가 직렬 1개씩.
실행 — ParallelACExecutor.execute_parallel: 레벨은 위→아래 직렬, 레벨 안 AC는 anyio.create_task_group으로 병렬.
결과 처리 — 매 메시지 진행 이벤트(progress/tool_called/heartbeat) 방출, 레벨 끝마다 충돌 감지·중재, 종료 시 OrchestratorResult 반환.
실제 예시
단일 AC를 실제 실행시키는 핵심 메시지 루프의 골자: async for로 스트림을 소비하며 메시지마다 stall 데드라인을 리셋(900s 침묵=stall)하고, resume_handle을 갱신하며, 30s마다 하트비트를 얹는다.
펼쳐보기: 실제 소스 발췌 ( _execute_atomic_ac + build_task_prompt)
# parallel_executor.py (_execute_atomic_ac, 5479~)with anyio.CancelScope( deadline=anyio.current_time() + STALL_TIMEOUT_SECONDS, # 900s(15분) 침묵 = stall) as stall_scope: async for message in self._adapter.execute_task( prompt=prompt, tools=tools, system_prompt=system_prompt, resume_handle=runtime_handle, ): # 메시지마다 stall 데드라인 리셋 (RC6 핵심) stall_scope.deadline = anyio.current_time() + STALL_TIMEOUT_SECONDS if message.resume_handle is not None: runtime_handle = self._remember_ac_runtime_handle(ac_index, message.resume_handle, ...) if runtime_handle is not None and runtime_handle.native_session_id: ac_session_id = runtime_handle.native_session_id messages.append(message); message_count += 1 # RC1: 메시지 흐름에 하트비트를 얹어 살아있음 알림 now = time.monotonic() if now - last_heartbeat >= HEARTBEAT_INTERVAL_SECONDS: # 30s heartbeat_event = create_heartbeat_event(session_id=session_id, ac_index=ac_index, ...)
# runner.py (build_task_prompt, 323~)def build_task_prompt(seed, strategy=None): if strategy is None: strategy = get_strategy(seed.task_type) ac_list = "\n".join(f"{i + 1}. {ac}" for i, ac in enumerate(seed.acceptance_criteria)) suffix = strategy.get_task_prompt_suffix() return f"""Execute the following task according to the acceptance criteria:## Goal{seed.goal}## Acceptance Criteria{ac_list}{render_auto_recursion_guard()}{suffix}"""
직접 만들 때 참고할 최소 골격(개념 재현용):
펼쳐보기: 최소 오케스트레이터 루프 골격 (개념 재현용)
async def run_orchestration(seed, adapter, event_store, max_workers=3): exec_id = f"exec_{uuid4().hex[:12]}" # 1) Seed → 프롬프트 system_prompt = build_system_prompt(seed) # 전략+계약+AC추적+복구 task_prompt = build_task_prompt(seed) # Goal + 번호매긴 AC # 2) AC 의존성 분석 → 레벨 그래프 (LLM 질의, 실패 시 단일 병렬 레벨 폴백) graph = await analyzer.analyze(seed.acceptance_criteria) plan = graph.to_execution_plan() # StagedExecutionPlan failed, blocked = set(), set() level_contexts = [] for stage in plan.stages: # 스테이지 직렬 executable = [i for i in stage.ac_indices if not any(d in failed or d in blocked for d in plan.get_dependencies(i))] async with anyio.create_task_group() as tg: # 스테이지 내부 병렬 results = [] for ac_idx in executable[:max_workers]: tg.start_soon(run_one_ac, ac_idx, adapter, system_prompt, level_contexts, results) # 레벨 리뷰 게이트: 파일 충돌 감지·중재 conflicts = coordinator.detect_file_conflicts(results) if conflicts: review = await coordinator.run_review(exec_id, conflicts, ctx, stage.index+1) level_contexts.append(make_context(results, review)) return OrchestratorResult(success=all_ok(results), session_id=..., execution_id=exec_id)async def run_one_ac(ac_idx, adapter, system_prompt, contexts, out): prompt = build_atomic_prompt(ac_idx, contexts) # 내 AC + 형제 경계 + 직전 컨텍스트 msgs, handle = [], None with anyio.CancelScope(deadline=anyio.current_time() + 900) as stall: # stall 감시 async for m in adapter.execute_task(prompt=prompt, tools=TOOLS, system_prompt=system_prompt, resume_handle=handle): stall.deadline = anyio.current_time() + 900 # 메시지마다 리셋 if m.resume_handle: handle = m.resume_handle msgs.append(m) # 30s마다 heartbeat, N개마다 progress 이벤트 방출 out.append(ac_result_from(ac_idx, msgs))
직접 구현 체크리스트:
Seed의 goal+acceptance_criteria를 번호 매긴 task 프롬프트로 변환했는가
system 프롬프트에 전략 fragment/계약/AC 추적/복구 프로토콜을 합쳤는가
AC 의존성을 그래프화·위상정렬(스테이지)했는가 (분석 실패 폴백=단일 병렬 레벨)
스테이지는 직렬, 스테이지 내 AC는 병렬(task group)로 돌리는가
선행 AC 실패/차단 시 의존 AC를 blocked로 건너뛰는가
execute_task를 async for로 소비하며 매 메시지 resume_handle을 갱신하는가