# 복합업무 임시팀(Composite Team) 시스템 설계서

> 버전: 1.0 (v0.4 — 에이전트 미팅 5사이클 최종 합의)
> 작성일: 2026-03-25
> 작업 ID: task-1045.1
> 작성자: 헤르메스(개발1팀장)
> 검토자: 불칸(백엔드), 아테나(UX), 아르고스(QA), 로키(DA/보안)

---

## 1. 문제 정의

현재 시스템의 한계:
- **개발팀**: 1팀=1봇 고정매핑. 개발 작업은 정상 동작.
- **논리적 팀(단일)**: 마케팅/디자인/출판이 가용 봇에서 단독 구동. 단일 전문 작업은 정상.
- **복합업무**: 마케팅+디자인처럼 2개 이상 논리적 팀 역량이 Phase별로 필요한 경우 → 방법 없음.
  - 아누가 수동으로 Phase 1(마케팅) → Phase 2(디자인) 순차 위임
  - 비효율적 + 중간 핸드오프에서 맥락 손실 발생

## 2. 해결책

하나의 봇 세션에 복합업무 임시팀을 구성하여, Opus 임시팀장이 Phase 관리/핸드오프/품질 검수를 일괄 수행.

### 기존 chain_manager.py 대비 장점
- 단일 세션 내 맥락 보존 (세션 간 맥락 손실 제거)
- AI가 Phase 관리를 자동 수행 (아누 수동 위임 불필요)
- Phase 간 레이턴시 최소화

## 3. CLI 인터페이스

```bash
python3 dispatch.py --composite marketing,design --task-file /path/to/task.md [--level normal|critical|security]
```

- `--team`과 `--composite`: `add_mutually_exclusive_group(required=True)`
- `--composite` 값: 쉼표 구분, 2개 이상 ~ MAX_COMPOSITE_TEAMS(3) 이하
- 허용 팀: COMPOSITE_ALLOWED_TEAMS (marketing, design, consulting, publishing)

## 4. 파일 구조

### 4-1. 신규 파일
- `utils/composite_constants.py`: 공유 상수 (SSOT)
- 테스트: 기존 test_dispatch.py, test_team_prompts.py에 클래스 추가

### 4-2. 수정 파일
- `dispatch.py`: composite 분기 추가
- `prompts/team_prompts.py`: build_composite_prompt() 추가

## 5. utils/composite_constants.py (SSOT)

```python
"""복합업무 임시팀 공유 상수 (Single Source of Truth)"""

COMPOSITE_ALLOWED_TEAMS = frozenset({"marketing", "design", "consulting", "publishing"})
MAX_COMPOSITE_TEAMS = 3

HANDOFF_REQUIRED_FIELDS: dict = {
    frozenset({"marketing", "design"}): [
        "플랫폼", "사이즈/형식", "톤앤매너", "CTA 문구", "타겟 오디언스",
    ],
    frozenset({"consulting", "publishing"}): [
        "대상 독자", "법적 검토 여부", "인용 출처", "데이터 정확성",
    ],
    frozenset({"marketing", "publishing"}): [
        "톤앤매너", "타겟 오디언스", "키워드", "SEO 메타",
    ],
    frozenset({"marketing", "consulting"}): [
        "타겟 고객", "상품 정보", "규제 사항", "CTA 문구",
    ],
}

DEFAULT_HANDOFF_FIELDS = ["핵심 결정사항", "산출물 파일 경로", "다음 Phase 요구사항"]
```

## 6. dispatch.py 변경사항

### 6-1. 임포트 추가
```python
from utils.composite_constants import (
    COMPOSITE_ALLOWED_TEAMS,
    MAX_COMPOSITE_TEAMS,
)
```

### 6-2. _validate_composite_teams()
```python
def _validate_composite_teams(teams_str: str) -> List[str]:
    """composite teams 문자열 파싱 + 유효성 검증"""
    teams = [t.strip() for t in teams_str.split(",") if t.strip()]
    if len(teams) < 2:
        raise ValueError(f"composite는 2개 이상의 팀이 필요합니다 (입력: {teams_str!r})")
    if len(teams) > MAX_COMPOSITE_TEAMS:
        raise ValueError(f"최대 {MAX_COMPOSITE_TEAMS}개 팀까지 허용 (입력: {len(teams)}개)")
    unknown = [t for t in teams if t not in COMPOSITE_ALLOWED_TEAMS]
    if unknown:
        raise ValueError(f"알 수 없는 팀 ID: {unknown}. 허용 목록(소문자): {sorted(COMPOSITE_ALLOWED_TEAMS)}")
    if len(teams) != len(set(teams)):
        raise ValueError(f"중복 팀 ID: {teams}")
    return teams
```

### 6-3. _validate_and_setup_task()
```python
def _validate_and_setup_task(
    task_id: str,
    task_desc: str,
    team_ids_for_check: List[str],  # composite: 포함 팀 리스트, 단일: [team_id]
    level: str,
    force: bool = False,
) -> Optional[dict]:
    """dispatch()와 _dispatch_composite() 공통 안전장치.

    포함 로직:
    - injection_guard 검사 (task_desc)
    - approval 검사
    - model_router 검사
    - task-timer start
    - 동일 팀 running 중복 체크 (팀 리스트 각각 순회, force 지원)

    비포함 (호출자 책임):
    - phases/chain_id 관련 로직
    - project-map 갱신
    - task_desc 파일 저장

    Returns: 에러 dict 또는 None(정상)
    """
```

### 6-4. dispatch() 시그니처 변경
```python
def dispatch(
    team_id: Optional[str] = None,    # str → Optional[str]
    task_desc: str = "",
    level: str = "normal",
    composite_teams: Optional[List[str]] = None,  # 신규
    ...
) -> dict:
    # XOR 검증
    if not team_id and not composite_teams:
        return {"status": "error", "message": "team_id 또는 composite_teams 중 하나 필수"}
    if team_id and composite_teams:
        return {"status": "error", "message": "team_id와 composite_teams는 동시 사용 불가"}
    # task_id 생성
    if task_id_param is None:
        task_id = generate_task_id()
    # composite 분기
    if composite_teams:
        return _dispatch_composite(composite_teams, task_desc, level, task_id=task_id, force=force, ...)
    # 기존 단일 팀 로직 (리팩터링: _validate_and_setup_task 호출)
    ...
```

### 6-5. _dispatch_composite()
```python
def _dispatch_composite(
    composite_teams: List[str],
    task_desc: str,
    level: str = "normal",
    task_id: Optional[str] = None,
    force: bool = False,
) -> dict:
    """복합업무 임시팀 디스패치.

    흐름:
    1. task_id 확인/생성
    2. _validate_composite_teams() 재검증
    3. injection_guard (composite 전용: high/critical → 즉시 차단)
    4. _validate_and_setup_task() 공통 안전장치
    5. _find_available_bot() 가용 봇 선택
    6. task_desc 파일 저장 + task_id prefix 치환
    7. build_composite_prompt() 호출
    8. _patch_timer_metadata(task_id, role="composite", composite_teams=composite_teams)
    9. cokacdir --cron 호출 (선택된 봇 키)
    10. 성공: set_bot_status, daily 로그, 반환
    11. 실패: _cleanup_task, 에러 반환
    """
```

### 6-6. _patch_timer_metadata() 타입 변경
```python
def _patch_timer_metadata(task_id: str, **metadata: Any) -> None:  # str → Any
```

### 6-7. main() argparse 변경
```python
# 기존 --team을 mutually_exclusive_group으로 변경
team_or_composite = parser.add_mutually_exclusive_group(required=True)
team_or_composite.add_argument("--team", choices=[...])
team_or_composite.add_argument("--composite", help="쉼표 구분 논리적 팀 ID 목록 (2~3개)")
```

### 6-8. 보안: injection_guard 이원화
```python
# _dispatch_composite() 내부
if _INJECTION_GUARD_AVAILABLE and _scan_content is not None:
    _scan_result = _scan_content(task_desc)
    if not _scan_result.is_safe:
        _high_threats = [t for t in _scan_result.threats if t.severity in ("high", "critical")]
        if _high_threats:
            # composite: 즉시 차단 (단일 팀과 다름)
            _cleanup_task(task_id)
            return {"status": "error", "message": f"보안 위협 감지 (composite 차단): {[t.pattern_name for t in _high_threats]}"}
```

## 7. team_prompts.py 변경사항

### 7-1. 임포트 추가
```python
from utils.composite_constants import (
    COMPOSITE_ALLOWED_TEAMS,
    DEFAULT_HANDOFF_FIELDS,
    HANDOFF_REQUIRED_FIELDS,
)
```

### 7-2. _load_logical_team_agents() (캐시 포함)
```python
_TEAM_AGENT_CACHE: dict = {}

def _load_logical_team_agents(team_id: str) -> dict:
    """org-details JSON에서 팀 에이전트 정보 로드 (캐시 포함)"""
    if team_id in _TEAM_AGENT_CACHE:
        return _TEAM_AGENT_CACHE[team_id]
    detail_path = Path(WORKSPACE_ROOT) / "memory" / "org-details" / f"{team_id}-team.json"
    if detail_path.exists():
        with open(detail_path, "r", encoding="utf-8") as f:
            result = json.load(f)
            _TEAM_AGENT_CACHE[team_id] = result
            return result
    # fallback
    fallback = {"members": TEAM_INFO.get(team_id, {}).get("members", "")}
    _TEAM_AGENT_CACHE[team_id] = fallback
    return fallback
```

### 7-3. build_composite_prompt() (독립 진입점)
```python
def build_composite_prompt(
    composite_teams: List[str],
    task_id: str,
    task_desc: str,
    level: str = "normal",
    project_id: Optional[str] = None,
) -> str:
    """복합업무 임시팀 프롬프트 생성 (독립 진입점).

    Defense in Depth: 내부에서 composite_teams 재검증.
    """
    # 내부 재검증
    for t in composite_teams:
        if t not in COMPOSITE_ALLOWED_TEAMS:
            raise ValueError(f"build_composite_prompt: 허용되지 않은 팀 ID: {t}")
    if len(composite_teams) < 2:
        raise ValueError("build_composite_prompt: 2개 이상의 팀 필요")

    # task_desc 파일 저장
    task_file_path = f"{WORKSPACE_ROOT}/memory/tasks/{task_id}.md"
    Path(task_file_path).parent.mkdir(parents=True, exist_ok=True)
    Path(task_file_path).write_text(task_desc, encoding="utf-8")

    # 각 팀 에이전트 로드
    team_agents = {team: _load_logical_team_agents(team) for team in composite_teams}

    # 핸드오프 필수 필드
    team_set = frozenset(composite_teams)
    handoff_fields = HANDOFF_REQUIRED_FIELDS.get(team_set, DEFAULT_HANDOFF_FIELDS)

    # 프롬프트 조립
    prompt = _assemble_composite_prompt(
        composite_teams, team_agents, handoff_fields,
        task_id, task_file_path, level, project_id,
    )

    # level 헤더
    if level == "critical":
        prompt = "**[CRITICAL] 이 작업은 중요도 critical입니다. 품질 우선으로 신중하게 작업하세요.**\n\n" + prompt
    elif level == "security":
        prompt = "**[SECURITY] 이 작업은 보안 중요 작업입니다. 보안 최우선으로 작업하세요.**\n\n" + prompt

    return prompt
```

### 7-4. _assemble_composite_prompt()
```python
def _assemble_composite_prompt(
    composite_teams: List[str],
    team_agents: dict,
    handoff_fields: list,
    task_id: str,
    task_file_path: str,
    level: str,
    project_id: Optional[str],
) -> str:
    """복합업무 프롬프트 조립.

    9개 섹션 구조:
    1. 임시팀장 페르소나 (고정 템플릿)
    2. 작업 지시
    3. 팀별 에이전트 목록 + 스킬
    4. Phase 관리 프로토콜
    5. 핸드오프 규격
    6. Quality Gate 규칙
    7. 보고서 규칙 (2레이어)
    8. 토큰 관리 지시
    9. 완료 마무리

    토큰 예산: 2팀 ~1,860 토큰 / 3팀 ~2,400 토큰 (4,000 상한 이내)
    """
    teams_str = ", ".join(composite_teams)
    report_path = f"{WORKSPACE_ROOT}/memory/reports/{task_id}.md"

    # 섹션 1: 임시팀장 페르소나 (고정 템플릿)
    s1 = (
        f"당신은 {teams_str} 복합업무 임시팀장입니다.\n"
        f"여러 논리적 팀의 역량을 Phase별로 조율하여 하나의 복합업무를 완수합니다.\n\n"
    )

    # 섹션 2: 작업 지시
    s2 = (
        f"## 작업 지시\n"
        f"- 작업 ID: {task_id}\n"
        f"- 작업 상세는 {task_file_path}를 읽고 파악하세요.\n\n"
    )

    # 섹션 3: 팀별 에이전트 목록
    s3 = "## 팀별 에이전트\n"
    for team_id in composite_teams:
        info = team_agents.get(team_id, {})
        s3 += f"\n### {team_id}\n"
        members = info.get("members_detail", info.get("members", {}))
        if isinstance(members, dict):
            for mid, minfo in members.items():
                name = minfo.get("name", mid)
                role = minfo.get("role", "")
                model = minfo.get("model", "sonnet")
                skill = minfo.get("mapped_skill", "")
                s3 += f"- **{name}** ({role}): model={model}"
                if skill:
                    s3 += f", skill={skill}"
                s3 += "\n"
        elif isinstance(members, str):
            s3 += f"- {members}\n"
    s3 += "\n"

    # 섹션 4: Phase 관리 프로토콜
    s4 = (
        f"## Phase 관리 프로토콜\n\n"
        f"### Phase 0: 계획 수립 (FYI — 감사 추적용)\n"
        f"1. task.md를 분석하여 Phase 계획을 수립하세요.\n"
        f"2. Phase 계획을 `{WORKSPACE_ROOT}/memory/tasks/{task_id}-phase-plan.md`에 저장하세요.\n"
        f"3. 각 Phase에 어떤 팀이 어떤 순서로 작업하는지 명시하세요.\n"
        f"4. Phase 계획 파일에 shell 명령, 파일 삭제, 시스템 경로 변경 지시를 포함하지 마세요.\n"
        f"5. 계획 수립 후 즉시 Phase 1을 시작하세요 (한정승인).\n\n"
        f"### Phase N (1~M): 순차 실행\n"
        f"1. 해당 Phase 팀의 에이전트만 활성화 (Task tool 사용)\n"
        f"2. 산출물을 파일로 저장\n"
        f"3. Quality Gate 검증 (아래 규칙 참조)\n"
        f"4. 핸드오프 문서 작성 (아래 규격 참조)\n\n"
        f"### Phase Final: 통합 보고서\n"
        f"모든 Phase 완료 후 통합 보고서를 {report_path}에 작성하세요.\n\n"
    )

    # 섹션 5: 핸드오프 규격
    fields_str = "\n".join(f"- [{f}]: (해당 내용 기재)" for f in handoff_fields)
    s5 = (
        f"## Phase 간 핸드오프 규격\n"
        f"각 Phase 완료 시 `{WORKSPACE_ROOT}/memory/tasks/{task_id}-handoff-{{N}}.md`를 작성하세요:\n\n"
        f"```markdown\n"
        f"## 핵심 요약 (200자 이내)\n"
        f"(다음 Phase 팀이 이 요약만 읽어도 작업 가능하도록 핵심만 기재)\n\n"
        f"## 산출물 파일\n"
        f"- (파일 경로): (설명)\n\n"
        f"## 필수 전달 사항\n"
        f"{fields_str}\n\n"
        f"## 다음 Phase 요구사항\n"
        f"- (구체적 요구사항)\n"
        f"```\n\n"
        f"⚠️ 다음 Phase에서는 핸드오프 파일의 '핵심 요약' 섹션만 참조하세요 (토큰 절감).\n\n"
    )

    # 섹션 6: Quality Gate
    s6 = (
        f"## Quality Gate 규칙\n"
        f"- 각 Phase 산출물에 대해 품질 검증을 수행하세요.\n"
        f"- 미충족 시 재작업 지시 (최대 2회)\n"
        f"- 2회 재작업 후에도 미충족 시:\n"
        f"  1. 이슈를 보고서에 기록\n"
        f"  2. 다음 Phase로 진행\n"
        f"  3. ⚠️ 이 상황은 반드시 보고서의 Executive Summary에 명시\n\n"
    )

    # 섹션 7: 보고서 규칙 (2레이어)
    s7 = (
        f"## 보고서 규칙 (2레이어)\n"
        f"보고서 경로: {report_path}\n\n"
        f"### 구조:\n"
        f"1. **Executive Summary** (최상단, 300자 이내)\n"
        f"   - 전체 결과: 성공/부분성공/실패\n"
        f"   - QG 실패 건수 및 Phase\n"
        f"   - 최종 산출물 파일 경로\n"
        f"   - 후속 조치 필요 여부\n\n"
        f"2. **Phase별 상세**\n"
        f"   - 각 Phase: S(상황), C(문제), Q(질문), A(답변)\n"
        f"   - Quality Gate 통과/실패 이력\n\n"
    )

    # 섹션 8: 토큰 관리
    s8 = (
        f"## 토큰 관리\n"
        f"- Phase 전환 시 이전 Phase 산출물 전문을 컨텍스트에 올리지 마세요.\n"
        f"- 핸드오프 파일의 '핵심 요약' 섹션만 참조하세요.\n"
        f"- 상세 내용이 필요하면 파일 경로를 기록하고 필요 시에만 Read하세요.\n\n"
    )

    # 섹션 9: 완료 마무리
    s9 = (
        f"## 완료 마무리\n"
        f"모든 Phase 완료 및 보고서 작성 후:\n"
        f"`bash {WORKSPACE_ROOT}/scripts/finish-task.sh {task_id}`\n"
    )

    return s1 + s2 + s3 + s4 + s5 + s6 + s7 + s8 + s9
```

## 8. 임시팀장 프로토콜

### 8-1. Phase 0: 계획 수립 (FYI)
- 목적: 감사 추적 + 비동기 알림
- 한정승인 작업이므로 아누 승인 대기 없이 즉시 실행
- Phase 계획 파일 생성 후 injection_guard 검사 지시 포함

### 8-2. Phase N: 순차 실행
- 해당 Phase 팀 에이전트 활성화 (Task tool, subagent_type=general-purpose)
- 산출물 파일 저장
- Quality Gate (최대 2회 재작업)
- 핸드오프 문서 생성

### 8-3. Phase Final: 통합 보고서
- 2레이어: Executive Summary + Phase별 상세
- finish-task.sh 실행

## 9. 보안 설계

### injection_guard 이원화
- 단일 팀 dispatch: 로깅만 (기존 호환)
- composite dispatch: high/critical → 즉시 error 반환

### Defense in Depth
- Layer 1 (CLI): _validate_composite_teams() in dispatch.py
- Layer 2 (Library): build_composite_prompt() 내부 재검증 in team_prompts.py
- Layer 3 (Prompt): 임시팀장에게 금지 패턴 명시

### 데이터 흐름 보안
- composite_teams 파라미터: 화이트리스트 검증
- AI 생성 중간 파일: injection_guard 검사 지시 (프롬프트 레벨, 강제력 제한적)

## 10. 대시보드 연동

### task-timers.json 확장 필드
- `role`: "composite"
- `composite_teams`: ["marketing", "design"]
- `current_phase`: "Phase 1: marketing" (임시팀장이 프롬프트 내에서 업데이트 불가 → 대시보드 표시용으로만 초기값 설정)

### 하위 호환성
- 기존 코드는 .get() 방식으로 필드 접근 → 새 필드 존재 여부에 무관하게 동작

## 11. 테스트 전략

### T1: _validate_composite_teams() 단위 테스트 (5+ 케이스)
- 정상 2팀, 정상 3팀, 빈 문자열, 연속 쉼표, 대소문자, 단일 팀, 4팀(MAX 초과), 중복, 존재하지 않는 팀

### T2: build_composite_prompt() 단위 테스트
- mock: _load_logical_team_agents, ANU_KEY
- 2팀/3팀 조합, level별 헤더, 핸드오프 필드 동적 생성

### T3: _dispatch_composite() 에러 경로
- 봇 없음, 잘못된 팀, cokacdir 실패, build_composite_prompt 예외

### T4: HANDOFF_REQUIRED_FIELDS
- 등록된 조합 → 전용 필드, 미등록 조합 → DEFAULT, 3팀 조합 → DEFAULT

### T5: _validate_and_setup_task()
- injection_guard, running 체크, timer start, cleanup

### T6: 상수 검증
- COMPOSITE_ALLOWED_TEAMS 존재 + TEAM_BOT 오염 방지

### T7: 기존 dispatch() 회귀 방지
- 기존 테스트 전체 통과 확인

## 12. 구현 순서

```
1. utils/composite_constants.py 생성
2. dispatch.py: _patch_timer_metadata **metadata: Any
3. dispatch.py: _validate_composite_teams() 추가
4. dispatch.py: _validate_and_setup_task() 추출
8. team_prompts.py: _load_logical_team_agents() + 캐시
9. team_prompts.py: build_composite_prompt() + _assemble_composite_prompt()
5. dispatch.py: _dispatch_composite() 구현
6. dispatch.py: dispatch() 시그니처 변경 + composite 분기
7. dispatch.py: main() argparse 변경
10. 테스트 작성
```

---

## 부록: 미팅 이력

- 사이클 1: 23개 개선점 (초기 설계 검토)
- 사이클 2: 15개 개선점 (보안/아키텍처 심화)
- 사이클 3: 10개 개선점 (구현 세부)
- 사이클 4: 5개 개선점 (최종 검증)
- 사이클 5: 4/4 합격 (최종 합의)
- 전체 미팅 기록: /home/jay/workspace/memory/meetings/composite-team-design.md