Files

T

DongHyeonka 6c53ded9cb init: llm-wiki-haness 하네스 설계

2026-07-24 14:21:35 +09:00

54 KiB

Raw Permalink Blame History

Deep-Research → Codex/Antigravity CLI 이식 Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Claude Code의 deep-research 하네스를 codex-cli·antigravity-cli 각각이 그대로 수행하도록 단일 외부 Python 드라이버로 이식한다.

Architecture: 결정론 제어 로직(dedup·예산·3-vote 정족수·랭킹)을 core.py 순수함수로 1:1 이식하고, 플랫폼 차이는 backends/(codex/antigravity) 어댑터로 흡수한다. 에이전트 실행기는 각 CLI의 headless 모드(codex exec / agy -p)이고 웹 조사는 각 CLI 네이티브 도구가 수행한다. MockBackend로 네트워크 0 상태에서 충실도를 단위/통합 테스트한다.

Tech Stack: Python 3.11+, Pydantic v2, asyncio/subprocess(stdlib), pytest, pytest-asyncio. 외부 검색 API·SDK 없음(CLI 구독으로 충당).

Spec: docs/superpowers/specs/2026-06-09-deep-research-codex-antigravity-port-design.md 원본 JS(충실도 기준): ~/.claude/projects/-home-donghyeon-dev-llm-wiki-private/88afa9ca-1e45-4353-9ff0-6361812d9053/workflows/scripts/deep-research-wf_aecef33f-4cc.js

File Structure

파일	책임
`scripts/deep-research/pyproject.toml`	패키지 메타·deps
`scripts/deep-research/deep_research/config.py`	상수(원본 동일) + 동시성 cap
`scripts/deep-research/deep_research/schemas.py`	5 Pydantic 모델 + JSON Schema export
`scripts/deep-research/deep_research/prompts.py`	6단계 프롬프트(원본 문자열 이식)
`scripts/deep-research/deep_research/core.py`	결정론 순수함수: norm_url/Deduper/rank_claims/tally/stats/degenerate
`scripts/deep-research/deep_research/pipeline.py`	asyncio 오케스트레이션(무배리어 pipeline / verify 배리어)
`scripts/deep-research/deep_research/backends/base.py`	AgentBackend ABC
`scripts/deep-research/deep_research/backends/mock.py`	테스트 더블
`scripts/deep-research/deep_research/backends/codex.py`	`codex exec --json --output-schema`
`scripts/deep-research/deep_research/backends/antigravity.py`	`agy -p` + 검증·재시도
`scripts/deep-research/deep_research/report.py`	Report → markdown/JSON
`scripts/deep-research/deep_research/__main__.py`	CLI 진입점
`scripts/deep-research/tests/test_core.py`	core 순수함수 단위테스트
`scripts/deep-research/tests/test_pipeline_mock.py`	MockBackend 통합 + 퇴화 경로 3종
`scripts/deep-research/README.md`	사용법·인증 전제

공통 규칙: LLM이 채우는 스키마 필드는 원본 JSON 키와 동일하게 camelCase 속성명(sourceQuality, publishDate, counterSource, openQuestions, sourceUrl)을 쓴다 — 프롬프트 지시·원본 wire 포맷과 1:1 일치시키기 위함. core 내부 dict도 동일 키 사용.

Task 0: 프로젝트 스캐폴드

Files:

Create: scripts/deep-research/pyproject.toml
Create: scripts/deep-research/deep_research/__init__.py
Create: scripts/deep-research/deep_research/backends/__init__.py
Create: scripts/deep-research/tests/__init__.py
Step 1: pyproject.toml 작성

[project]
name = "deep-research-driver"
version = "0.1.0"
description = "Port of Claude Code deep-research harness to Codex/Antigravity CLIs"
requires-python = ">=3.11"
dependencies = ["pydantic>=2.6"]

[project.optional-dependencies]
dev = ["pytest>=8", "pytest-asyncio>=0.23"]

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]

[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"

Step 2: 빈 __init__.py 3개 생성 (deep_research/, deep_research/backends/, tests/) — 내용 없음.
Step 3: 가상환경·설치 검증

Run: cd scripts/deep-research && python -m venv .venv && .venv/bin/pip install -e ".[dev]" Expected: 설치 성공, pytest 사용 가능.

Step 4: Commit

git add scripts/deep-research/pyproject.toml scripts/deep-research/deep_research/__init__.py scripts/deep-research/deep_research/backends/__init__.py scripts/deep-research/tests/__init__.py
git commit -m "chore(deep-research): scaffold python driver package"

Task 1: config.py (상수)

Files:

Create: scripts/deep-research/deep_research/config.py
Step 1: config 작성 (원본 상수 1:1)

"""원본 deep-research-wf JS 와 동일한 상수. 변경 금지(충실도)."""
from dataclasses import dataclass


@dataclass(frozen=True)
class Config:
    VOTES_PER_CLAIM: int = 3
    REFUTATIONS_REQUIRED: int = 2
    MAX_FETCH: int = 15
    MAX_VERIFY_CLAIMS: int = 25
    CONCURRENCY: int = 10           # asyncio.Semaphore (원본 Workflow cap 대응)
    ANTIGRAVITY_RETRIES: int = 2    # 스키마 검증 실패 시 재프롬프트 횟수


DEFAULT = Config()

Step 2: import 검증

Run: cd scripts/deep-research && .venv/bin/python -c "from deep_research.config import DEFAULT; print(DEFAULT.MAX_FETCH)" Expected: 15

Step 3: Commit

git add scripts/deep-research/deep_research/config.py
git commit -m "feat(deep-research): add config constants (1:1 with JS harness)"

Task 2: schemas.py (5 Pydantic 모델)

Files:

Create: scripts/deep-research/deep_research/schemas.py
Test: scripts/deep-research/tests/test_core.py (이 태스크에서 스키마 검증 테스트만 먼저)
Step 1: 스키마 검증 실패 테스트 작성

tests/test_core.py 에 추가:

import pytest
from pydantic import ValidationError
from deep_research.schemas import Scope, Angle, Search, SearchResult, Extract, Verdict, Report


def test_scope_requires_min_3_angles():
    with pytest.raises(ValidationError):
        Scope(question="q", summary="s", angles=[Angle(label="a", query="x")])


def test_scope_accepts_5_angles():
    angles = [Angle(label=f"a{i}", query="x") for i in range(5)]
    s = Scope(question="q", summary="s", angles=angles)
    assert len(s.angles) == 5


def test_search_relevance_enum_enforced():
    with pytest.raises(ValidationError):
        SearchResult(url="http://a", title="t", relevance="bogus")


def test_json_schema_export_works():
    schema = Scope.model_json_schema()
    assert "angles" in schema["properties"]

Step 2: 테스트 실패 확인

Run: cd scripts/deep-research && .venv/bin/pytest tests/test_core.py -k "scope or relevance or json_schema" -v Expected: FAIL — ModuleNotFoundError: deep_research.schemas

Step 3: schemas.py 구현

"""원본 5 SCHEMA 의 Pydantic v2 포팅. LLM-facing 필드는 원본 JSON 키(camelCase) 유지."""
from typing import Literal, Optional
from pydantic import BaseModel, Field


class Angle(BaseModel):
    label: str
    query: str
    rationale: Optional[str] = None


class Scope(BaseModel):
    question: str
    summary: str
    angles: list[Angle] = Field(min_length=3, max_length=6)


class SearchResult(BaseModel):
    url: str
    title: str
    snippet: Optional[str] = None
    relevance: Literal["high", "medium", "low"]


class Search(BaseModel):
    results: list[SearchResult] = Field(max_length=6)


class Claim(BaseModel):
    claim: str
    quote: str
    importance: Literal["central", "supporting", "tangential"]


class Extract(BaseModel):
    sourceQuality: Literal["primary", "secondary", "blog", "forum", "unreliable"]
    publishDate: Optional[str] = None
    claims: list[Claim] = Field(max_length=5)


class Verdict(BaseModel):
    refuted: bool
    evidence: str
    confidence: Literal["high", "medium", "low"]
    counterSource: Optional[str] = None


class Finding(BaseModel):
    claim: str
    confidence: Literal["high", "medium", "low"]
    sources: list[str]
    evidence: str
    vote: Optional[str] = None


class Report(BaseModel):
    summary: str
    findings: list[Finding]
    caveats: str
    openQuestions: Optional[list[str]] = None

Step 4: 테스트 통과 확인

Run: cd scripts/deep-research && .venv/bin/pytest tests/test_core.py -k "scope or relevance or json_schema" -v Expected: PASS (4 passed)

Step 5: Commit

git add scripts/deep-research/deep_research/schemas.py scripts/deep-research/tests/test_core.py
git commit -m "feat(deep-research): add 5 Pydantic schemas (1:1 with JS SCHEMA)"

Task 3: prompts.py (6단계 프롬프트 이식)

Files:

Create: scripts/deep-research/deep_research/prompts.py

원본 JS 의 프롬프트 문자열을 그대로 이식한다. 함수는 인자를 받아 문자열을 만든다.

Step 1: prompts.py 구현 (원본 문자열 1:1)

"""원본 deep-research-wf JS 의 프롬프트 문자열 이식. 표현 변경 금지(충실도)."""


def scope_prompt(question: str) -> str:
    return (
        "Decompose this research question into complementary search angles.\n\n"
        "## Question\n" + question + "\n\n"
        "## Task\n"
        "Generate 5 distinct web search queries that together cover the question from "
        "different angles. Pick angles that suit the question's domain. Examples:\n"
        "- broad/primary  · academic/technical  · recent news  · contrarian/skeptical  · practitioner/implementation\n"
        "- For medical: anatomy · common causes · serious differentials · authoritative refs · red flags\n"
        "- For tech: state-of-art · benchmarks · limitations · industry adoption · cost/tradeoffs\n\n"
        "Make queries specific enough to surface high-signal results. Avoid redundancy.\n"
        "Return: the question (verbatim or lightly normalized), a 1-2 sentence decomposition "
        "strategy, and the angles.\n\nStructured output only."
    )


def search_prompt(question: str, angle) -> str:
    return (
        "## Web Searcher: " + angle.label + "\n\n"
        'Research question: "' + question + '"\n\n'
        "Your angle: **" + angle.label + "** — " + (angle.rationale or "") + "\n"
        "Search query: `" + angle.query + "`\n\n"
        "## Task\nUse web search with the query above (or a refined version). Return the top "
        "4-6 most relevant results.\n"
        "Rank by relevance to the ORIGINAL question, not just the search query. Skip obvious "
        "SEO spam/content farms.\n"
        "Include a short snippet capturing why each result is relevant.\n\nStructured output only."
    )


def fetch_prompt(question: str, source: dict, angle: str) -> str:
    return (
        "## Source Extractor\n\n"
        'Research question: "' + question + '"\n\n'
        "Fetch and extract key claims from this source:\n"
        "**URL:** " + source["url"] + "\n**Title:** " + source["title"] + "\n**Found via:** " + angle + " search\n\n"
        "## Task\n1. Fetch the page content.\n"
        "2. Assess source quality: primary research/institution? secondary reporting? blog/opinion? forum? unreliable?\n"
        "3. Extract 2-5 FALSIFIABLE claims that bear on the research question. Each claim must:\n"
        "   - be a concrete, checkable statement (not vague generalities)\n"
        "   - include a direct quote from the source as support\n"
        "   - be rated central/supporting/tangential to the research question\n"
        "4. Note publish date if available.\n\n"
        'If the fetch fails or the page is irrelevant/paywalled, return claims: [] and '
        'sourceQuality: "unreliable".\n\nStructured output only.'
    )


def verify_prompt(question: str, claim: dict, v: int, votes_per_claim: int, refutations_required: int) -> str:
    return (
        "## Adversarial Claim Verifier (voter " + str(v + 1) + "/" + str(votes_per_claim) + ")\n\n"
        "Be SKEPTICAL. Try to REFUTE this claim. ≥" + str(refutations_required) + "/" + str(votes_per_claim) + " refutations kill it.\n\n"
        "## Research question\n" + question + "\n\n"
        '## Claim under review\n"' + claim["claim"] + '"\n\n'
        "**Source:** " + claim["sourceUrl"] + " (" + claim["sourceQuality"] + ")\n"
        '**Supporting quote:** "' + claim["quote"] + '"\n\n'
        "## Checklist\n"
        "1. Is the claim actually supported by the quote, or is it an overreach/misread?\n"
        "2. Search for contradicting evidence — does any credible source dispute or heavily qualify this?\n"
        "3. Is the source quality sufficient for the claim's strength? (extraordinary claims need primary sources)\n"
        "4. Is the claim outdated? (check dates — old claims about fast-moving fields are suspect)\n"
        "5. Is this a marketing claim / press release / cherry-picked benchmark / forum speculation?\n\n"
        "**refuted=true** if: unsupported by quote / contradicted / low-quality source for strong claim / outdated / marketing fluff.\n"
        "**refuted=false** ONLY if: claim is well-supported, current, and source quality matches claim strength.\n"
        "Default to refuted=true if uncertain.\n\nStructured output only. Evidence MUST be specific."
    )


def synth_prompt(question: str, block: str, killed_block: str, confirmed_count: int, votes_per_claim: int) -> str:
    return (
        "## Synthesis: research report\n\n"
        "**Question:** " + question + "\n\n"
        + str(confirmed_count) + " claims survived " + str(votes_per_claim) + "-vote adversarial verification. "
        "Merge semantic duplicates and synthesize.\n\n"
        "## Confirmed claims\n" + block + "\n" + killed_block + "\n\n"
        "## Instructions\n"
        "1. Identify claims that say the same thing — merge them, combine their sources.\n"
        "2. Group related claims into coherent findings. Each finding should directly address the research question.\n"
        "3. Assign confidence per finding: high (multiple primary sources, unanimous votes), medium (secondary sources or split votes), low (single source or blog-quality).\n"
        "4. Write a 3-5 sentence executive summary answering the research question.\n"
        "5. Note caveats: what's uncertain, what sources were weak, what time-sensitivity applies.\n"
        "6. List 2-4 open questions that emerged but weren't answered.\n\nStructured output only."
    )

Step 2: import 검증

Run: cd scripts/deep-research && .venv/bin/python -c "from deep_research import prompts; print(prompts.scope_prompt('q')[:20])" Expected: Decompose this resea

Step 3: Commit

git add scripts/deep-research/deep_research/prompts.py
git commit -m "feat(deep-research): port 6-phase prompt strings from JS harness"

Task 4: core — norm_url + Deduper (TDD)

Files:

Create: scripts/deep-research/deep_research/core.py
Test: scripts/deep-research/tests/test_core.py (추가)
Step 1: 실패 테스트 작성 — tests/test_core.py 에 추가:

from deep_research.core import norm_url, Deduper
from deep_research.config import Config


def test_norm_url_strips_www_and_trailing_slash():
    assert norm_url("http://www.Example.com/Path/") == "example.com/path"


def test_norm_url_bare_host():
    assert norm_url("https://example.com") == "example.com"


def test_norm_url_fallback_on_garbage():
    assert norm_url("not a url") == "not a url"


def test_deduper_filters_exact_dupes():
    d = Deduper(Config(MAX_FETCH=15))
    r = [{"url": "http://a.com/x", "title": "t", "relevance": "high"}]
    assert len(d.filter_novel("angle1", r)) == 1
    # same normalized url from another angle -> dup
    r2 = [{"url": "http://www.a.com/x/", "title": "t2", "relevance": "high"}]
    assert d.filter_novel("angle2", r2) == []
    assert len(d.dupes) == 1


def test_deduper_budget_drops_medium_low_when_slots_exhausted():
    d = Deduper(Config(MAX_FETCH=1))
    first = [{"url": "http://a.com/1", "title": "t", "relevance": "high"}]
    d.filter_novel("a1", first)  # consumes the only slot
    more = [
        {"url": "http://b.com/2", "title": "t", "relevance": "medium"},
        {"url": "http://c.com/3", "title": "t", "relevance": "low"},
    ]
    assert d.filter_novel("a2", more) == []
    assert len(d.budget_dropped) == 2


def test_deduper_high_passes_even_when_slots_exhausted():
    # 원본: high(rank 0)는 budget 조건(rank>=1)에 안 걸려 slot<=0 이어도 통과
    d = Deduper(Config(MAX_FETCH=1))
    d.filter_novel("a1", [{"url": "http://a.com/1", "title": "t", "relevance": "high"}])
    high = [{"url": "http://d.com/4", "title": "t", "relevance": "high"}]
    assert len(d.filter_novel("a2", high)) == 1
    assert d.fetch_slots == -1

Step 2: 테스트 실패 확인

Run: cd scripts/deep-research && .venv/bin/pytest tests/test_core.py -k "norm_url or deduper" -v Expected: FAIL — ImportError: cannot import name 'norm_url'

Step 3: core.py 구현 (1차)

"""결정론 제어 로직 — 원본 JS 와 1:1. 네트워크·LLM 의존 0."""
from urllib.parse import urlparse
from deep_research.config import Config

REL_RANK = {"high": 0, "medium": 1, "low": 2}
IMP_RANK = {"central": 0, "supporting": 1, "tangential": 2}
QUAL_RANK = {"primary": 0, "secondary": 1, "blog": 2, "forum": 3, "unreliable": 4}


def norm_url(u: str) -> str:
    try:
        p = urlparse(u)
        host = (p.hostname or "")
        if not host:
            return u.lower()
        if host.startswith("www."):
            host = host[4:]
        path = (p.path or "").rstrip("/")
        return (host + path).lower()
    except Exception:
        return u.lower()


def host_of(u: str) -> str:
    try:
        h = urlparse(u).hostname or "unknown"
        return h[4:] if h.startswith("www.") else h
    except Exception:
        return "unknown"


class Deduper:
    """원본 pipeline stage-2 의 dedup+budget 로직. 호출 순서에 결정론적."""

    def __init__(self, config: Config):
        self.seen: dict[str, dict] = {}
        self.dupes: list[dict] = []
        self.budget_dropped: list[dict] = []
        self.fetch_slots = config.MAX_FETCH

    def filter_novel(self, angle: str, results: list[dict]) -> list[dict]:
        ordered = sorted(results, key=lambda r: REL_RANK[r["relevance"]])
        novel: list[dict] = []
        for r in ordered:
            key = norm_url(r["url"])
            if key in self.seen:
                self.dupes.append({**r, "angle": angle, "dupOf": self.seen[key]})
                continue
            if self.fetch_slots <= 0 and REL_RANK[r["relevance"]] >= 1:
                self.budget_dropped.append({**r, "angle": angle})
                continue
            self.seen[key] = {"angle": angle, "title": r["title"]}
            self.fetch_slots -= 1
            novel.append(r)
        return novel

Step 4: 테스트 통과 확인

Run: cd scripts/deep-research && .venv/bin/pytest tests/test_core.py -k "norm_url or deduper" -v Expected: PASS (6 passed)

Step 5: Commit

git add scripts/deep-research/deep_research/core.py scripts/deep-research/tests/test_core.py
git commit -m "feat(deep-research): core norm_url + Deduper (TDD, 1:1 with JS)"

Task 5: core — rank_claims + tally (TDD, 정족수 엣지 고정)

Files:

Modify: scripts/deep-research/deep_research/core.py
Test: scripts/deep-research/tests/test_core.py (추가)
Step 1: 실패 테스트 작성 — tests/test_core.py 에 추가:

from deep_research.core import rank_claims, tally


def _claim(imp, qual, name="c"):
    return {"claim": name, "importance": imp, "sourceQuality": qual}


def test_rank_claims_orders_by_importance_then_quality():
    claims = [
        _claim("tangential", "primary", "t-prim"),
        _claim("central", "blog", "c-blog"),
        _claim("central", "primary", "c-prim"),
    ]
    ranked = rank_claims(claims, max_verify=25)
    assert [c["claim"] for c in ranked] == ["c-prim", "c-blog", "t-prim"]


def test_rank_claims_truncates_to_max():
    claims = [_claim("central", "primary", f"c{i}") for i in range(30)]
    assert len(rank_claims(claims, max_verify=25)) == 25


def test_tally_survives_2_valid_0_refute():
    v = [{"refuted": False}, {"refuted": False}, {"refuted": False}]
    t = tally(v, votes_per_claim=3, refutations_required=2)
    assert t["survives"] is True and t["refutedVotes"] == 0


def test_tally_killed_2_refute():
    v = [{"refuted": True}, {"refuted": True}, {"refuted": False}]
    t = tally(v, votes_per_claim=3, refutations_required=2)
    assert t["survives"] is False and t["refutedVotes"] == 2


def test_tally_all_abstain_does_not_survive():
    # ⚠️ 거짓 생존 차단: all-None -> refuted=0 이지만 valid<2 라 미생존
    t = tally([None, None, None], votes_per_claim=3, refutations_required=2)
    assert t["survives"] is False and t["abstained"] == 3


def test_tally_one_valid_two_abstain_does_not_survive():
    t = tally([{"refuted": False}, None, None], votes_per_claim=3, refutations_required=2)
    assert t["survives"] is False  # valid(1) < 2

Step 2: 테스트 실패 확인

Run: cd scripts/deep-research && .venv/bin/pytest tests/test_core.py -k "rank_claims or tally" -v Expected: FAIL — ImportError: cannot import name 'rank_claims'

Step 3: core.py 에 추가

def rank_claims(claims: list[dict], max_verify: int) -> list[dict]:
    return sorted(
        claims,
        key=lambda c: (IMP_RANK[c["importance"]], QUAL_RANK[c["sourceQuality"]]),
    )[:max_verify]


def tally(verdicts: list, votes_per_claim: int, refutations_required: int) -> dict:
    """원본 verify 정족수 산식 1:1. None=기권."""
    valid = [v for v in verdicts if v is not None]
    refuted = sum(1 for v in valid if v["refuted"])
    abstained = votes_per_claim - len(valid)
    survives = len(valid) >= refutations_required and refuted < refutations_required
    return {
        "valid": valid,
        "refutedVotes": refuted,
        "abstained": abstained,
        "survives": survives,
    }

Step 4: 테스트 통과 확인

Run: cd scripts/deep-research && .venv/bin/pytest tests/test_core.py -k "rank_claims or tally" -v Expected: PASS (6 passed)

Step 5: Commit

git add scripts/deep-research/deep_research/core.py scripts/deep-research/tests/test_core.py
git commit -m "feat(deep-research): core rank_claims + tally with abstention edge (TDD)"

Task 6: core — synthesis 블록 빌더 + stats + 퇴화 경로 빌더 (TDD)

Files:

Modify: scripts/deep-research/deep_research/core.py
Test: scripts/deep-research/tests/test_core.py (추가)

원본의 synthesis block/killedBlock 문자열 조립과 최종 stats·퇴화 반환을 순수 함수로 분리.

Step 1: 실패 테스트 작성 — tests/test_core.py 에 추가:

from deep_research.core import build_synth_blocks, build_stats


def test_build_synth_blocks_includes_vote_and_source():
    confirmed = [{
        "claim": "X causes Y", "quote": "q", "sourceUrl": "http://a", "sourceQuality": "primary",
        "valid": [{"refuted": False, "confidence": "high", "evidence": "e"}],
        "refutedVotes": 0,
    }]
    killed = [{
        "claim": "Z", "sourceUrl": "http://b", "valid": [{"refuted": True}], "refutedVotes": 1,
    }]
    block, killed_block = build_synth_blocks(confirmed, killed)
    assert "X causes Y" in block and "1-0" in block
    assert "Refuted claims" in killed_block and "Z" in killed_block


def test_build_synth_blocks_no_killed():
    block, killed_block = build_synth_blocks([{
        "claim": "X", "quote": "q", "sourceUrl": "http://a", "sourceQuality": "primary",
        "valid": [{"refuted": False, "confidence": "high", "evidence": "e"}], "refutedVotes": 0,
    }], [])
    assert killed_block == ""


def test_build_stats_agent_calls_formula():
    s = build_stats(angles=5, sources=12, claims=20, voted=18, confirmed=10, killed=8,
                    after_synth=6, dupes=3, budget_dropped=2, votes_per_claim=3)
    # 1 + angles + sources + voted*votes_per_claim + 1
    assert s["agentCalls"] == 1 + 5 + 12 + 18 * 3 + 1
    assert s["confirmed"] == 10 and s["afterSynthesis"] == 6

Step 2: 테스트 실패 확인

Run: cd scripts/deep-research && .venv/bin/pytest tests/test_core.py -k "synth_blocks or build_stats" -v Expected: FAIL — ImportError

Step 3: core.py 에 추가

CONF_RANK = {"high": 0, "medium": 1, "low": 2}


def _vote_str(c: dict) -> str:
    return str(len(c["valid"]) - c["refutedVotes"]) + "-" + str(c["refutedVotes"])


def build_synth_blocks(confirmed: list[dict], killed: list[dict]) -> tuple[str, str]:
    """원본 synthesize 의 block / killedBlock 문자열 조립."""
    parts = []
    for i, c in enumerate(confirmed):
        non_refuted = [v for v in c["valid"] if not v["refuted"]]
        best = sorted(non_refuted, key=lambda v: CONF_RANK[v["confidence"]])[0]
        parts.append(
            "### [" + str(i) + "] " + c["claim"] + "\n"
            + "Vote: " + _vote_str(c) + " · Source: " + c["sourceUrl"] + " (" + c["sourceQuality"] + ")\n"
            + 'Quote: "' + c["quote"] + '"\nVerifier evidence (' + best["confidence"] + "): " + best["evidence"] + "\n"
        )
    block = "\n".join(parts)

    if killed:
        killed_block = "\n## Refuted claims (for transparency)\n" + "\n".join(
            '- "' + c["claim"] + '" (' + c["sourceUrl"] + ", vote " + _vote_str(c) + ")"
            for c in killed
        )
    else:
        killed_block = ""
    return block, killed_block


def build_stats(*, angles, sources, claims, voted, confirmed, killed,
                after_synth, dupes, budget_dropped, votes_per_claim) -> dict:
    return {
        "angles": angles,
        "sourcesFetched": sources,
        "claimsExtracted": claims,
        "claimsVerified": voted,
        "confirmed": confirmed,
        "killed": killed,
        "afterSynthesis": after_synth,
        "urlDupes": dupes,
        "budgetDropped": budget_dropped,
        "agentCalls": 1 + angles + sources + (voted * votes_per_claim) + 1,
    }

Step 4: 테스트 통과 확인

Run: cd scripts/deep-research && .venv/bin/pytest tests/test_core.py -k "synth_blocks or build_stats" -v Expected: PASS (3 passed)

Step 5: 전체 core 테스트 회귀 확인

Run: cd scripts/deep-research && .venv/bin/pytest tests/test_core.py -v Expected: PASS (이전 태스크 포함 전부)

Step 6: Commit

git add scripts/deep-research/deep_research/core.py scripts/deep-research/tests/test_core.py
git commit -m "feat(deep-research): core synth-block builder + stats (TDD)"

Task 7: backends/base.py + mock.py

Files:

Create: scripts/deep-research/deep_research/backends/base.py
Create: scripts/deep-research/deep_research/backends/mock.py
Step 1: base.py 구현

from abc import ABC, abstractmethod
from pydantic import BaseModel


class AgentBackend(ABC):
    """run_agent: 프롬프트를 1개 CLI headless 에이전트로 실행, schema 로 검증된 객체 반환.
    실패/사용자-skip -> None (원본 .filter(Boolean) 의미)."""

    @abstractmethod
    async def run_agent(self, prompt: str, schema: type[BaseModel], *, label: str) -> BaseModel | None:
        ...

Step 2: mock.py 구현

from pydantic import BaseModel
from deep_research.backends.base import AgentBackend


class MockBackend(AgentBackend):
    """테스트 더블. label prefix -> 응답객체(또는 None) 매핑. 네트워크·subprocess 0.

    responses 의 키는 label prefix. 가장 먼저 매칭되는 prefix 의 값을 반환.
    값이 콜러블이면 (prompt, label) 로 호출해 동적 응답 가능."""

    def __init__(self, responses: dict, default=None):
        self.responses = responses
        self.default = default
        self.calls: list[str] = []

    async def run_agent(self, prompt: str, schema: type[BaseModel], *, label: str):
        self.calls.append(label)
        for prefix, resp in self.responses.items():
            if label.startswith(prefix):
                return resp(prompt, label) if callable(resp) else resp
        return self.default

Step 3: import 검증

Run: cd scripts/deep-research && .venv/bin/python -c "from deep_research.backends.mock import MockBackend; print('ok')" Expected: ok

Step 4: Commit

git add scripts/deep-research/deep_research/backends/base.py scripts/deep-research/deep_research/backends/mock.py
git commit -m "feat(deep-research): AgentBackend ABC + MockBackend test double"

Task 8: pipeline.py (asyncio 오케스트레이션, MockBackend TDD)

Files:

Create: scripts/deep-research/deep_research/pipeline.py
Test: scripts/deep-research/tests/test_pipeline_mock.py
Step 1: happy-path + 퇴화 경로 실패 테스트 작성 — tests/test_pipeline_mock.py:

import pytest
from deep_research.pipeline import run_research
from deep_research.backends.mock import MockBackend
from deep_research.config import Config
from deep_research.schemas import Scope, Angle, Search, SearchResult, Extract, Claim, Verdict, Report

CFG = Config(MAX_FETCH=15, MAX_VERIFY_CLAIMS=25)


def _scope():
    return Scope(question="Q", summary="s",
                 angles=[Angle(label=f"a{i}", query=f"q{i}") for i in range(3)])


def _search(prompt, label):
    # 각 각도마다 고유 URL 1개
    n = label.split(":")[1]
    return Search(results=[SearchResult(url=f"http://{n}.com/x", title="t", relevance="high")])


def _extract(prompt, label):
    return Extract(sourceQuality="primary",
                   claims=[Claim(claim="C-" + label, quote="q", importance="central")])


def _verdict_pass(prompt, label):
    return Verdict(refuted=False, evidence="e", confidence="high")


def _report():
    return Report(summary="done", findings=[], caveats="none")


async def test_happy_path_returns_report_and_stats():
    backend = MockBackend({
        "scope": _scope(),
        "search:": _search,
        "fetch:": _extract,
        "v": _verdict_pass,
        "synthesize": _report(),
    })
    out = await run_research("Q", backend, config=CFG)
    assert out["summary"] == "done"
    assert out["stats"]["confirmed"] == 3   # 3 각도 × 1 claim, 전부 생존
    assert out["stats"]["angles"] == 3


async def test_empty_question_returns_error():
    out = await run_research("   ", MockBackend({}), config=CFG)
    assert "error" in out


async def test_no_claims_degenerate():
    backend = MockBackend({
        "scope": _scope(),
        "search:": _search,
        "fetch:": lambda p, l: Extract(sourceQuality="unreliable", claims=[]),
    })
    out = await run_research("Q", backend, config=CFG)
    assert out["findings"] == [] and out["stats"]["claims"] == 0


async def test_all_refuted_degenerate():
    backend = MockBackend({
        "scope": _scope(),
        "search:": _search,
        "fetch:": _extract,
        "v": lambda p, l: Verdict(refuted=True, evidence="e", confidence="high"),
    })
    out = await run_research("Q", backend, config=CFG)
    assert out["findings"] == [] and out["stats"]["confirmed"] == 0
    assert len(out["refuted"]) == 3


async def test_synth_failure_salvages_confirmed():
    backend = MockBackend({
        "scope": _scope(),
        "search:": _search,
        "fetch:": _extract,
        "v": _verdict_pass,
        "synthesize": None,   # 합성 실패
    })
    out = await run_research("Q", backend, config=CFG)
    assert out["findings"] == [] and len(out["confirmed"]) == 3

Step 2: 테스트 실패 확인

Run: cd scripts/deep-research && .venv/bin/pytest tests/test_pipeline_mock.py -v Expected: FAIL — ModuleNotFoundError: deep_research.pipeline

Step 3: pipeline.py 구현

"""asyncio 오케스트레이션. 원본 pipeline(무배리어)/parallel(배리어) 의미 재현."""
import asyncio
from deep_research.config import Config
from deep_research.core import Deduper, host_of, rank_claims, tally, build_synth_blocks, build_stats
from deep_research.schemas import Scope, Search, Extract, Verdict, Report
from deep_research import prompts


async def run_research(question: str, backend, *, config: Config) -> dict:
    q = (question or "").strip()
    if not q:
        return {"error": "No research question provided."}

    scope = await backend.run_agent(prompts.scope_prompt(q), Scope, label="scope")
    if scope is None:
        return {"error": "Scope agent returned no result — cannot decompose the question."}

    deduper = Deduper(config)
    sem = asyncio.Semaphore(config.CONCURRENCY)
    dedup_lock = asyncio.Lock()

    async def search_and_fetch(angle) -> list[dict]:
        async with sem:
            sr = await backend.run_agent(
                prompts.search_prompt(q, angle), Search, label="search:" + angle.label)
        if sr is None:
            return []
        async with dedup_lock:  # 공유 상태(seen/fetch_slots) 임계구역 직렬화
            novel = deduper.filter_novel(angle.label, [r.model_dump() for r in sr.results])

        async def fetch_one(source: dict):
            async with sem:
                ext = await backend.run_agent(
                    prompts.fetch_prompt(q, source, angle.label),
                    Extract, label="fetch:" + host_of(source["url"]))
            if ext is None:
                return None
            return {
                "url": source["url"], "title": source["title"], "angle": angle.label,
                "sourceQuality": ext.sourceQuality, "publishDate": ext.publishDate,
                "claims": [{**c.model_dump(), "sourceUrl": source["url"], "sourceQuality": ext.sourceQuality}
                           for c in ext.claims],
            }

        fetched = await asyncio.gather(*[fetch_one(s) for s in novel])
        return [f for f in fetched if f is not None]

    per_angle = await asyncio.gather(*[search_and_fetch(a) for a in scope.angles])
    all_sources = [s for sub in per_angle for s in sub]
    all_claims = [c for s in all_sources for c in s["claims"]]
    ranked = rank_claims(all_claims, config.MAX_VERIFY_CLAIMS)

    def _sources_out():
        return [{"url": s["url"], "quality": s["sourceQuality"], "angle": s["angle"],
                 "claimCount": len(s["claims"])} for s in all_sources]

    if not ranked:
        return {
            "question": q,
            "summary": f"No claims extracted. {len(all_sources)} sources fetched, all empty/failed.",
            "findings": [], "refuted": [], "sources": _sources_out(),
            "stats": {"angles": len(scope.angles), "sources": len(all_sources),
                      "claims": 0, "dupes": len(deduper.dupes)},
        }

    # ── Verify (배리어) ──
    async def verify_claim(claim: dict) -> dict:
        async def one_vote(v: int):
            async with sem:
                return await backend.run_agent(
                    prompts.verify_prompt(q, claim, v, config.VOTES_PER_CLAIM, config.REFUTATIONS_REQUIRED),
                    Verdict, label="v" + str(v) + ":" + claim["claim"][:40])
        verdicts = await asyncio.gather(*[one_vote(v) for v in range(config.VOTES_PER_CLAIM)])
        t = tally([vd.model_dump() if vd is not None else None for vd in verdicts],
                  config.VOTES_PER_CLAIM, config.REFUTATIONS_REQUIRED)
        return {**claim, **t}

    voted = await asyncio.gather(*[verify_claim(c) for c in ranked])
    confirmed = [c for c in voted if c["survives"]]
    killed = [c for c in voted if not c["survives"]]

    def _refuted_out():
        return [{"claim": c["claim"],
                 "vote": str(len(c["valid"]) - c["refutedVotes"]) + "-" + str(c["refutedVotes"]),
                 "source": c["sourceUrl"]} for c in killed]

    if not confirmed:
        return {
            "question": q,
            "summary": f"All {len(voted)} claims refuted by adversarial verification. Research inconclusive.",
            "findings": [], "refuted": _refuted_out(), "sources": _sources_out(),
            "stats": build_stats(angles=len(scope.angles), sources=len(all_sources),
                                 claims=len(all_claims), voted=len(voted), confirmed=0,
                                 killed=len(killed), after_synth=0, dupes=len(deduper.dupes),
                                 budget_dropped=len(deduper.budget_dropped),
                                 votes_per_claim=config.VOTES_PER_CLAIM),
        }

    # ── Synthesize ──
    block, killed_block = build_synth_blocks(confirmed, killed)
    report = await backend.run_agent(
        prompts.synth_prompt(q, block, killed_block, len(confirmed), config.VOTES_PER_CLAIM),
        Report, label="synthesize")

    if report is None:
        return {
            "question": q,
            "summary": f"Synthesis step was skipped or failed — returning {len(confirmed)} verified claims unmerged.",
            "findings": [],
            "confirmed": [{"claim": c["claim"], "source": c["sourceUrl"], "quote": c["quote"],
                           "vote": str(len(c["valid"]) - c["refutedVotes"]) + "-" + str(c["refutedVotes"])}
                          for c in confirmed],
            "refuted": _refuted_out(), "sources": _sources_out(),
            "stats": build_stats(angles=len(scope.angles), sources=len(all_sources),
                                 claims=len(all_claims), voted=len(voted), confirmed=len(confirmed),
                                 killed=len(killed), after_synth=0, dupes=len(deduper.dupes),
                                 budget_dropped=len(deduper.budget_dropped),
                                 votes_per_claim=config.VOTES_PER_CLAIM),
        }

    return {
        "question": q,
        **report.model_dump(),
        "refuted": _refuted_out(), "sources": _sources_out(),
        "stats": build_stats(angles=len(scope.angles), sources=len(all_sources),
                             claims=len(all_claims), voted=len(voted), confirmed=len(confirmed),
                             killed=len(killed), after_synth=len(report.findings),
                             dupes=len(deduper.dupes), budget_dropped=len(deduper.budget_dropped),
                             votes_per_claim=config.VOTES_PER_CLAIM),
    }

Step 4: 테스트 통과 확인

Run: cd scripts/deep-research && .venv/bin/pytest tests/test_pipeline_mock.py -v Expected: PASS (5 passed)

Step 5: 전체 회귀

Run: cd scripts/deep-research && .venv/bin/pytest -v Expected: PASS (전체)

Step 6: Commit

git add scripts/deep-research/deep_research/pipeline.py scripts/deep-research/tests/test_pipeline_mock.py
git commit -m "feat(deep-research): asyncio pipeline orchestration (MockBackend TDD, 4 degenerate paths)"

Task 9: report.py (Report → markdown/JSON)

Files:

Create: scripts/deep-research/deep_research/report.py
Test: scripts/deep-research/tests/test_pipeline_mock.py (추가)
Step 1: 실패 테스트 작성 — tests/test_pipeline_mock.py 에 추가:

from deep_research.report import to_markdown


def test_to_markdown_renders_summary_and_stats():
    result = {
        "question": "Q", "summary": "ans",
        "findings": [{"claim": "F1", "confidence": "high", "sources": ["http://a"], "evidence": "e"}],
        "caveats": "c", "refuted": [], "sources": [],
        "stats": {"angles": 5, "confirmed": 1},
    }
    md = to_markdown(result)
    assert "# Deep Research" in md and "ans" in md and "F1" in md

Step 2: 실패 확인

Run: cd scripts/deep-research && .venv/bin/pytest tests/test_pipeline_mock.py -k to_markdown -v Expected: FAIL — ModuleNotFoundError

Step 3: report.py 구현

import json


def to_markdown(result: dict) -> str:
    if "error" in result:
        return "# Deep Research\n\n**Error:** " + result["error"]

    lines = ["# Deep Research", "", "**Question:** " + result.get("question", ""), ""]
    lines += ["## Summary", result.get("summary", ""), ""]

    findings = result.get("findings", [])
    if findings:
        lines.append("## Findings")
        for i, f in enumerate(findings, 1):
            srcs = ", ".join(f.get("sources", []))
            lines += [
                f"### {i}. {f['claim']}  _({f['confidence']})_",
                f"{f.get('evidence', '')}",
                f"Sources: {srcs}",
                "",
            ]

    caveats = result.get("caveats")
    if caveats:
        lines += ["## Caveats", caveats, ""]

    oq = result.get("openQuestions")
    if oq:
        lines += ["## Open Questions"] + [f"- {q}" for q in oq] + [""]

    refuted = result.get("refuted", [])
    if refuted:
        lines.append("## Refuted (transparency)")
        for r in refuted:
            lines.append(f"- \"{r['claim']}\" (vote {r.get('vote', '')}, {r.get('source', '')})")
        lines.append("")

    lines += ["## Stats", "```json", json.dumps(result.get("stats", {}), ensure_ascii=False, indent=2), "```"]
    return "\n".join(lines)


def to_json(result: dict) -> str:
    return json.dumps(result, ensure_ascii=False, indent=2)

Step 4: 통과 확인

Run: cd scripts/deep-research && .venv/bin/pytest tests/test_pipeline_mock.py -k to_markdown -v Expected: PASS

Step 5: Commit

git add scripts/deep-research/deep_research/report.py scripts/deep-research/tests/test_pipeline_mock.py
git commit -m "feat(deep-research): report renderer (markdown + json)"

Task 10: backends/codex.py (`codex exec --json --output-schema`)

Files:

Create: scripts/deep-research/deep_research/backends/codex.py
Test: scripts/deep-research/tests/test_pipeline_mock.py (파서 단위테스트 추가; subprocess 는 모킹)

codex 의 JSONL stdout 에서 최종 agent 메시지(스키마 JSON)를 뽑는 파서를 분리해 단위테스트한다. 실제 codex exec 호출은 수동 스모크(Task 13).

Step 1: 파서 실패 테스트 작성 — tests/test_pipeline_mock.py 에 추가:

from deep_research.backends.codex import extract_final_json


def test_extract_final_json_from_jsonl():
    jsonl = "\n".join([
        '{"type":"reasoning","text":"thinking"}',
        '{"type":"web_search","query":"x"}',
        '{"type":"agent_message","text":"{\\"question\\":\\"Q\\",\\"summary\\":\\"s\\",\\"angles\\":[]}"}',
    ])
    obj = extract_final_json(jsonl)
    assert obj["summary"] == "s"


def test_extract_final_json_returns_none_when_absent():
    assert extract_final_json('{"type":"reasoning","text":"only"}') is None

Step 2: 실패 확인

Run: cd scripts/deep-research && .venv/bin/pytest tests/test_pipeline_mock.py -k extract_final_json -v Expected: FAIL — ModuleNotFoundError

Step 3: codex.py 구현

"""Codex CLI headless backend. `codex exec --json --output-schema` subprocess."""
import asyncio
import json
import os
import tempfile
from pydantic import BaseModel, ValidationError
from deep_research.backends.base import AgentBackend


def extract_final_json(jsonl_stdout: str) -> dict | None:
    """JSONL 이벤트 스트림에서 마지막 agent_message 의 text(JSON) 파싱."""
    final = None
    for line in jsonl_stdout.splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            evt = json.loads(line)
        except json.JSONDecodeError:
            continue
        if evt.get("type") in ("agent_message", "message") and "text" in evt:
            final = evt["text"]
    if final is None:
        return None
    try:
        return json.loads(final)
    except json.JSONDecodeError:
        return None


class CodexBackend(AgentBackend):
    def __init__(self, *, neutral_cwd: str | None = None, model: str | None = None):
        self.neutral_cwd = neutral_cwd or tempfile.gettempdir()
        self.model = model

    async def run_agent(self, prompt: str, schema: type[BaseModel], *, label: str):
        with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as fh:
            json.dump(schema.model_json_schema(), fh)
            schema_path = fh.name
        cmd = ["codex", "exec", "--json", "--output-schema", schema_path,
               "--cd", self.neutral_cwd, "-s", "read-only"]
        if self.model:
            cmd += ["-m", self.model]
        cmd.append(prompt)
        try:
            proc = await asyncio.create_subprocess_exec(
                *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
            out, _ = await proc.communicate()
        finally:
            os.unlink(schema_path)
        if proc.returncode != 0:
            return None
        obj = extract_final_json(out.decode("utf-8", "replace"))
        if obj is None:
            return None
        try:
            return schema.model_validate(obj)
        except ValidationError:
            return None

Step 4: 통과 확인

Run: cd scripts/deep-research && .venv/bin/pytest tests/test_pipeline_mock.py -k extract_final_json -v Expected: PASS (2 passed)

Step 5: Commit

git add scripts/deep-research/deep_research/backends/codex.py scripts/deep-research/tests/test_pipeline_mock.py
git commit -m "feat(deep-research): codex backend (codex exec --json --output-schema)"

Task 11: backends/antigravity.py (`agy -p` + 검증·재시도)

Files:

Create: scripts/deep-research/deep_research/backends/antigravity.py
Test: scripts/deep-research/tests/test_pipeline_mock.py (파서 단위테스트 추가)

agy 의 output-schema 플래그는 미확인(spec §11) → 프롬프트에 스키마 JSON 을 첨부하고, 출력에서 JSON 블록을 추출·검증·재시도. JSON 추출기를 분리해 단위테스트.

Step 1: 추출기 실패 테스트 작성 — tests/test_pipeline_mock.py 에 추가:

from deep_research.backends.antigravity import extract_json_block


def test_extract_json_block_from_fenced():
    text = 'prelude\n```json\n{"refuted": false, "evidence": "e", "confidence": "high"}\n```\ntrailing'
    obj = extract_json_block(text)
    assert obj["confidence"] == "high"


def test_extract_json_block_bare_object():
    obj = extract_json_block('noise {"refuted": true, "evidence": "e", "confidence": "low"} more')
    assert obj["refuted"] is True


def test_extract_json_block_none_when_absent():
    assert extract_json_block("no json here") is None

Step 2: 실패 확인

Run: cd scripts/deep-research && .venv/bin/pytest tests/test_pipeline_mock.py -k extract_json_block -v Expected: FAIL — ModuleNotFoundError

Step 3: antigravity.py 구현

"""Antigravity CLI headless backend. `agy -p` subprocess + 드라이버 스키마 검증·재시도."""
import asyncio
import json
import re
import tempfile
from pydantic import BaseModel, ValidationError
from deep_research.backends.base import AgentBackend

_FENCE = re.compile(r"```(?:json)?\s*(\{.*?\})\s*```", re.DOTALL)


def extract_json_block(text: str) -> dict | None:
    """출력 텍스트에서 JSON 객체 추출: 펜스 우선, 없으면 최외곽 중괄호."""
    m = _FENCE.search(text)
    candidates = [m.group(1)] if m else []
    if not candidates:
        start, depth = -1, 0
        for i, ch in enumerate(text):
            if ch == "{":
                if depth == 0:
                    start = i
                depth += 1
            elif ch == "}" and depth > 0:
                depth -= 1
                if depth == 0 and start >= 0:
                    candidates.append(text[start:i + 1])
                    break
    for c in candidates:
        try:
            return json.loads(c)
        except json.JSONDecodeError:
            continue
    return None


class AntigravityBackend(AgentBackend):
    def __init__(self, *, retries: int = 2, neutral_cwd: str | None = None, model: str | None = None):
        self.retries = retries
        self.neutral_cwd = neutral_cwd or tempfile.gettempdir()
        self.model = model

    async def _invoke(self, prompt: str) -> str | None:
        cmd = ["agy", "-p", prompt, "--cd", self.neutral_cwd]
        if self.model:
            cmd += ["-m", self.model]
        proc = await asyncio.create_subprocess_exec(
            *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
        out, _ = await proc.communicate()
        if proc.returncode != 0:
            return None
        return out.decode("utf-8", "replace")

    async def run_agent(self, prompt: str, schema: type[BaseModel], *, label: str):
        schema_json = json.dumps(schema.model_json_schema())
        full = (prompt + "\n\n## Output format\nReturn ONLY a JSON object conforming to this "
                "JSON Schema (no prose, no markdown fences):\n" + schema_json)
        for attempt in range(self.retries + 1):
            text = await self._invoke(full if attempt == 0 else
                                      full + "\n\nYour previous output was invalid. Return ONLY the JSON object.")
            if text is None:
                continue
            obj = extract_json_block(text)
            if obj is None:
                continue
            try:
                return schema.model_validate(obj)
            except ValidationError:
                continue
        return None

Step 4: 통과 확인

Run: cd scripts/deep-research && .venv/bin/pytest tests/test_pipeline_mock.py -k extract_json_block -v Expected: PASS (3 passed)

Step 5: Commit

git add scripts/deep-research/deep_research/backends/antigravity.py scripts/deep-research/tests/test_pipeline_mock.py
git commit -m "feat(deep-research): antigravity backend (agy -p + validate/retry)"

Task 12: main.py + README

Files:

Create: scripts/deep-research/deep_research/__main__.py
Create: scripts/deep-research/README.md
Step 1: __main__.py 구현

"""CLI: python -m deep_research --backend {codex|antigravity} "<질문>" [--json]"""
import argparse
import asyncio
import shutil
import sys
from deep_research.config import DEFAULT
from deep_research.pipeline import run_research
from deep_research.report import to_markdown, to_json


def _make_backend(name: str):
    if name == "codex":
        if shutil.which("codex") is None:
            sys.exit("error: `codex` CLI not found on PATH. Install + login (subscription).")
        from deep_research.backends.codex import CodexBackend
        return CodexBackend()
    if name == "antigravity":
        if shutil.which("agy") is None:
            sys.exit("error: `agy` CLI not found on PATH. Install + login (subscription).")
        from deep_research.backends.antigravity import AntigravityBackend
        return AntigravityBackend(retries=DEFAULT.ANTIGRAVITY_RETRIES)
    sys.exit(f"error: unknown backend '{name}'")


def main(argv=None):
    ap = argparse.ArgumentParser(prog="deep_research")
    ap.add_argument("question")
    ap.add_argument("--backend", required=True, choices=["codex", "antigravity"])
    ap.add_argument("--json", action="store_true", help="emit raw JSON instead of markdown")
    ap.add_argument("--concurrency", type=int, default=DEFAULT.CONCURRENCY)
    args = ap.parse_args(argv)

    backend = _make_backend(args.backend)
    cfg = DEFAULT.__class__(CONCURRENCY=args.concurrency)
    result = asyncio.run(run_research(args.question, backend, config=cfg))
    print(to_json(result) if args.json else to_markdown(result))


if __name__ == "__main__":
    main()

Step 2: README.md 작성

# deep-research driver

Claude Code 의 deep-research 하네스를 **codex-cli / antigravity-cli** 로 이식한 외부 Python 드라이버.

## 전제
- `codex` 또는 `agy` CLI 설치 + 로그인(구독). 별도 검색 API 키 불필요 — 웹 조사는 각 CLI 네이티브 도구가 수행.

## 사용
```bash
python -m deep_research --backend codex "2026년 한국 전기차 보조금 정책 변화" 
python -m deep_research --backend antigravity "..." --json

구조

core.py — 결정론 제어(dedup·예산·3-vote 정족수·랭킹). 원본 JS 와 1:1, 순수함수.
backends/ — 플랫폼 어댑터(codex/antigravity/mock).
충실도 기준 원본: deep-research-wf_aecef33f-4cc.js (spec 참조).

테스트

pip install -e ".[dev]" && pytest

MockBackend 로 네트워크 0 상태에서 6페이즈 + 퇴화 경로 3종 검증.


- [ ] **Step 3: CLI smoke (인자 파싱만, CLI 미설치 시 안내 확인)**

Run: `cd scripts/deep-research && .venv/bin/python -m deep_research --backend codex "test" 2>&1 | head -1` (codex 미설치면 안내 메시지)
Expected: 인자 파싱 동작 — codex 있으면 실행, 없으면 `error: codex CLI not found`.

- [ ] **Step 4: 전체 테스트 회귀**

Run: `cd scripts/deep-research && .venv/bin/pytest -v`
Expected: PASS (전체)

- [ ] **Step 5: Commit**

```bash
git add scripts/deep-research/deep_research/__main__.py scripts/deep-research/README.md
git commit -m "feat(deep-research): CLI entrypoint + README"

Task 13: 실 backend 스모크 (수동, 선택)

Files: 없음 (실행 검증만)

Step 1: codex 실 스모크

Run: cd scripts/deep-research && .venv/bin/python -m deep_research --backend codex "What are the main tradeoffs of virtual threads vs reactive programming in Java?" Expected: 보고서 markdown + stats. agentCalls 가 1+angles+sources+voted*3+1 와 부합. web_search 가 실제로 돌았는지 codex 로그로 확인.

Step 2: antigravity 실 스모크

Run: cd scripts/deep-research && .venv/bin/python -m deep_research --backend antigravity "<같은 질문>" Expected: 보고서 출력. 스키마 검증 실패율 높으면 --output format 플래그 등 agy 옵션 조사(spec §11 리스크), antigravity.py 재프롬프트 동작 확인.

Step 3: 관찰 기록

두 backend 산출물 차이(스키마 준수율, 속도, 웹도구 동작)를 docs/superpowers/notes/2026-06-09-deep-research-port-smoke.md 에 메모(Write 도구 사용).

Self-Review (작성자 체크)

1. Spec coverage

§3 모듈 구조 → Task 0~12 전부 매핑 ✓
§4 상수 → Task 1 ✓ · §5 스키마 → Task 2 ✓ · §6 6페이즈 → Task 3(프롬프트)+Task 8(흐름) ✓
§6.2 dedup 수학 → Task 4 ✓ · §6.4 rank → Task 5 ✓ · §6.5 정족수 abstention 엣지 → Task 5 ✓
§6.7 퇴화 경로 3종 → Task 8 테스트 ✓ · §6.8 stats → Task 6 ✓
§7 동시성(무배리어/배리어/dedup 락) → Task 8 ✓
§10 테스트(core 단위 + MockBackend 통합) → Task 4~8 ✓
backends(codex/antigravity/mock) → Task 7/10/11 ✓
§1.3 CLI headless + 네이티브 웹 → Task 10/11 ✓ (web 계층 없음 일관)

2. Placeholder scan — TBD/TODO 없음. 모든 코드 스텝에 실제 코드 포함 ✓

3. Type consistency

Config 필드명 Task1↔전 태스크 일치 ✓
core dict 키 sourceUrl/sourceQuality/refutedVotes/valid/survives 일관(Task5 tally 출력 ↔ Task6 build_synth_blocks 입력 ↔ Task8 pipeline) ✓
host_of(Task4) ↔ pipeline fetch label(Task8) 일치 ✓
backend run_agent(prompt, schema, *, label) 시그니처 base↔mock↔codex↔antigravity 일치 ✓
스키마 camelCase 필드(sourceQuality/publishDate/counterSource/openQuestions) ↔ 프롬프트 지시 일치 ✓

54 KiB Raw Permalink Blame History Unescape Escape

Deep-Research → Codex/Antigravity CLI 이식 Implementation Plan

File Structure

Task 0: 프로젝트 스캐폴드

Task 1: config.py (상수)

Task 2: schemas.py (5 Pydantic 모델)

Task 3: prompts.py (6단계 프롬프트 이식)

Task 4: core — norm_url + Deduper (TDD)

Task 5: core — rank_claims + tally (TDD, 정족수 엣지 고정)

Task 6: core — synthesis 블록 빌더 + stats + 퇴화 경로 빌더 (TDD)

Task 7: backends/base.py + mock.py

Task 8: pipeline.py (asyncio 오케스트레이션, MockBackend TDD)

Task 9: report.py (Report → markdown/JSON)

Task 10: backends/codex.py (codex exec --json --output-schema)

Task 11: backends/antigravity.py (agy -p + 검증·재시도)

Task 12: __main__.py + README

구조

테스트

Task 13: 실 backend 스모크 (수동, 선택)

Self-Review (작성자 체크)

54 KiB

Raw Permalink Blame History

Task 10: backends/codex.py (`codex exec --json --output-schema`)

Task 11: backends/antigravity.py (`agy -p` + 검증·재시도)

Task 12: main.py + README