ouroboros · 오케스트레이터 실행 루프(Seed→프롬프트→에이전트 실행)

한 줄 요약

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

[!note] 핵심 모양: “단일 호출”이 아니라 레벨 그래프를 위→아래 한 줄씩, 각 줄은 병렬로 도는 루프다. 세로는 직렬, 가로는 병렬.

핵심 정리

루프가 다루는 주요 데이터 구조 4종. 전체 스키마는 콜아웃으로 분산했다.

구조	한 줄 역할	근거
`OrchestratorResult`	루프의 최종 성적표(반환값)	runner.py:149
`AgentMessage`	에이전트가 흘리는 메시지 한 건	adapter.py:645
`ExecutionStage` / `StagedExecutionPlan`	레벨 그래프(직렬 묶음)	dependency_analyzer.py:118,137
`FileConflict` / `CoordinatorReview`	레벨 사이 충돌 검사·중재	coordinator.py:85,102

[!note]- 펼쳐보기: 전체 필드표 4종 (Result / Message / Stage / Conflict) OrchestratorResult (runner.py:149) — 루프의 최종 반환값

이름 타입 설명
success bool 전체 AC 충족 여부
session_id str 재개(resume)용 세션 식별자
execution_id str 워크플로 실행 ID (exec_<hex12>)
summary dict success/failure_count·total_levels 등
messages_processed int 총 메시지 수
final_message str 최종 결과 메시지
duration_seconds float 소요 시간

AgentMessage (adapter.py:645) — 한 건씩 소비하는 스트림 단위

이름 타입 설명
type str assistant/user/tool/result/system
content str 사람이 읽을 본문
tool_name str|None 도구 호출이면 도구 이름
data dict 부가(subtype==error면 에러, session_id 등)
resume_handle RuntimeHandle|None 재개 핸들(매 메시지 갱신)
is_final / is_error (prop) bool type=="result" / data["subtype"]=="error"

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,…] 적용된 수정 설명

이름	타입	설명
success	bool	전체 AC 충족 여부
session_id	str	재개(resume)용 세션 식별자
execution_id	str	워크플로 실행 ID (`exec_<hex12>`)
summary	dict	success/failure_count·total_levels 등
messages_processed	int	총 메시지 수
final_message	str	최종 결과 메시지
duration_seconds	float	소요 시간

이름	타입	설명
type	str	assistant/user/tool/result/system
content	str	사람이 읽을 본문
tool_name	str\|None	도구 호출이면 도구 이름
data	dict	부가(`subtype`==error면 에러, `session_id` 등)
resume_handle	RuntimeHandle\|None	재개 핸들(매 메시지 갱신)
`is_final` / `is_error` (prop)	bool	`type=="result"` / `data["subtype"]=="error"`

이름	타입	설명
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.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

[!note]- 펼쳐보기: 단계별 상세 (함수·폴백 규칙)

트리거 — 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마다 하트비트를 얹는다.

[!note]- 펼쳐보기: 실제 소스 발췌 (_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}
"""

직접 만들 때 참고할 최소 골격(개념 재현용):

[!note]- 펼쳐보기: 최소 오케스트레이터 루프 골격 (개념 재현용)

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))

직접 구현 체크리스트:

요약 & 셀프체크

3줄 요약:

Seed의 합격 기준들을 의존성 레벨로 나눠 레벨은 직렬·레벨 안은 병렬로 AI 에이전트를 돌린다.
각 에이전트는 자기 AC 한 개와 경계 안내만 받아 충돌을 줄이고, 레벨이 끝날 때마다 코디네이터가 파일 충돌을 검사·중재한다.
도는 동안 stall 타이머·하트비트·watchdog·취소 레지스트리가 멈춤과 폭주를 막고, 끝나면 OrchestratorResult로 성적표를 낸다.

스스로 답해보기:

“세로는 직렬, 가로는 병렬”은 레벨 그래프에서 정확히 무엇을 가리키는가?
코디네이터 리뷰 세션은 항상 도는가, 어떤 조건일 때만 도는가? 그 이유는?
AI가 15분 동안 아무 메시지도 안 보내면 무슨 일이 일어나며, 메시지가 올 때마다 리셋되는 것은 무엇인가?

연결

OB_개요 · _분석축_루브릭 · OB_20_spec-engine-seed-and-double-diamond(Seed가 어떻게 만들어지나) · OB_30_event-sourcing-and-projection-readmodel(진행 이벤트가 어디 쌓이나) · OB_50_provider-adapters-and-backend-neutral-runtime(execute_task 어댑터) · OB_10_entrypoint-cli-and-ooo-command-routing(ooo run 진입)

[!tip]- Codex 교차검증 메모 (근거 파일) 원본 분석 노트는 동작을 소스 라인 단위로 추적해 작성됐다. 핵심 사실은 모두 아래 근거에서 확인됐고, 재작성에서 사실은 바꾸지 않고 학습 흐름으로만 재구성했다.

…/orchestrator/runner.py — OrchestratorResult(149), build_system_prompt(290)/build_task_prompt(323), execute_seed(2012)/prepare_session(2057)/execute_precreated_session(2127), _execute_parallel(2707), 취소 레지스트리(211~261)

…/orchestrator/parallel_executor.py — execute_parallel(3658), 스테이지 직렬+AC 병렬 루프(3862~), 코디네이터 리뷰 게이트(4202~~4244), _execute_atomic_ac(5203)·execute_task 메시지 루프(5479~~5549), STALL_TIMEOUT 900s/HEARTBEAT 30s(1782)

…/orchestrator/coordinator.py — FileConflict(85)/CoordinatorReview(102)/LevelCoordinator(193), detect_file_conflicts(302), run_review(345), _build_review_prompt(496)

…/orchestrator/dependency_analyzer.py — ACNode(74)/ExecutionStage(118)/StagedExecutionPlan(137)/DependencyGraph(174), to_execution_plan(215), DEPENDENCY_ANALYSIS_PROMPT(232)

…/orchestrator/runtime_factory.py — create_agent_runtime(42): backend별(claude/codex/opencode/gemini/…) 런타임 선택

…/orchestrator/adapter.py — RuntimeHandle(405), AgentMessage(645)/is_final/is_error, execute_task(1225) 스트리밍 시그니처

…/runtime/watchdog.py — Watchdog(121)/WatchdogDecision(92), check()(169) 벽시계 초과 시 runtime.watchdog.cancel 1회 방출, replay 멱등성

…/runtime/controls.py — RuntimeControls(39): session_wall_clock_seconds(기본 4h), watchdog_enabled

…/.ouroboros/ — mechanical.toml(build/test/timeout), seeds/*.yaml (스펙주도 산출물)