🧬 技術調査 - ハーネスを実行時に進化させる Test-Time Harness Evolution

# 公開実装例: README の Install 手順
git clone https://github.com/junnie00/TTHE.git
cd TTHE
pip install -r requirements.txt

# 公開実装例: README の Configure 手順
cp config.example.yaml config.yaml
export OPENAI_API_KEY=sk-...

# 公開実装例: config.example.yaml の主要キー
llm:
  provider: openai
  base_url: https://api.deepseek.com/v1
  api_key_env: OPENAI_API_KEY
  solver_model: deepseek-chat
  controller_model: deepseek-chat
  temperature: 0.0
  request_timeout: 60

dataset:
  name: demo
  db_id: demo

output_dir: runs

# 公開実装例: 内蔵 demo DB の smoke test
python -m tthe.optimize --db demo --cap 5 --max-rounds 3

# 公開実装例: BIRD Mini-Dev を設定した上での実行
python -m tthe.optimize --db california_schools --batch-size 10 --max-rounds 3

TTHE/
├── tthe/
│   ├── optimize.py
│   ├── evolve.py
│   ├── proposer.py
│   ├── evaluator.py
│   ├── reward.py
│   ├── harness_base.py
│   ├── react_baseline.py
│   ├── bridge.py
│   ├── bt.py
│   ├── claude_wrapper.py
│   └── agents/
│       ├── bare.py
│       └── react.py
├── ase/
│   ├── llm.py
│   ├── dataset.py
│   ├── db.py
│   └── demo_db.py
├── config.example.yaml
├── requirements.txt
└── LICENSE

# 実装案/例: SQL 以外の agent に移植するための Harness interface
from typing import Protocol

class Harness(Protocol):
    id: str
    source_path: str

    def run(self, task_input) -> "ExecutionTrace":
        """モデル呼び出し、ツール呼び出し、検証、復旧を実行し、出力と trace を返す。"""
        ...

# 実装案/例: TTHE の trace を他ドメインに移植する schema
from dataclasses import dataclass
from typing import Any

@dataclass
class ExecutionTrace:
    prompts: list[str]
    completions: list[str]
    tool_calls: list[dict[str, Any]]
    stdout: str
    stderr: str
    artifacts: list[str]
    runtime_states: list[dict[str, Any]]
    errors: list[str]
    probes: list[dict[str, Any]]
    final_output: Any

# 実装案/例: TTHE loop のドメイン非依存疑似コード
def tthe_loop(committed_harness, batches, branches_g, rounds_r):
    for batch in batches:
        traces = observe(committed_harness, batch)
        branches = [clone(committed_harness) for _ in range(branches_g)]

        for round_index in range(1, rounds_r + 1):
            next_branches = []
            for branch_id, parent in enumerate(branches):
                child = propose(
                    parent=parent,
                    peer_traces=traces,
                    role=role_for(branch_id),
                )
                next_branches.append(child if is_valid(child) else parent)
            branches = next_branches
            traces.update(observe_all(branches, batch))

        committed_harness = judge(branches, traces, batch)
        report_score_after_commit(committed_harness, batch)

    return committed_harness

# 実装案/例: proxy は oracle ではなく evidence として返す
def compute_proxies(trace: ExecutionTrace) -> dict:
    return {
        "execution_health": trace.errors == [] and is_well_formed(trace.final_output),
        "round_trip_consistency": estimate_round_trip_match(trace),
        "public_test_pass_rate": run_public_tests(trace) if has_public_tests(trace) else None,
        "notes": "proxy is evidence, not correctness",
    }

# 公開実装例: API を使わない smoke test を意図する場合
# config.yaml で llm.provider: mock を設定してから実行する
python -m tthe.optimize --db demo --cap 5 --max-rounds 3

# 公開実装例: BIRD 用 DB を設定済みの場合
export BIRD_DEV_FILE=mini_dev.json
python -m tthe.optimize \
  --db california_schools \
  --batch-size 10 \
  --group 3 \
  --max-rounds 3 \
  --initial-harness react

# 公開実装例: README が示す Dataset interface
class Dataset:
    def get_database(self, db_id) -> Database:
        ...

    def eval_questions(self, db_id):
        ...

# 実装案/例: バッチごとの成果物ディレクトリ
runs/<run_id>/batch_0007/
├── committed_harness.py
├── candidate_harnesses/
├── traces.jsonl
├── proxies.json
├── judge_decision.md
├── post_commit_gold_score.json
└── rollback_pointer.txt

# 実装案/例: 固定 baseline の測定
python -m tthe.optimize \
  --db california_schools \
  --batch-size 10 \
  --group 1 \
  --max-rounds 1 \
  --initial-harness react

# 実装案/例: TTHE 進化 loop の測定
python -m tthe.optimize \
  --db california_schools \
  --batch-size 10 \
  --group 3 \
  --max-rounds 3 \
  --initial-harness react

{"event":"observe","batch_id":"B0007","harness_id":"H_t","n_inputs":10,"proxy":{"exec_success_rate":0.90,"public_pass_rate":0.62},"gold_used":false}
{"event":"propose","batch_id":"B0007","round":1,"branch":1,"parent_id":"H_t","child_id":"H_t_g1_r1","role":"conservative-repair","valid":true}
{"event":"propose","batch_id":"B0007","round":1,"branch":2,"parent_id":"H_t","child_id":"H_t_g2_r1","role":"exploration","valid":false,"fallback_to":"H_t"}
{"event":"judge","batch_id":"B0007","final_branches":["H_t_g1_r3","H_t_g2_r3","H_t_g3_r3"],"committed":"H_t_g3_r3","reexecution_probe":true,"gold_used":false}
{"event":"score","batch_id":"B0007","harness_id":"H_t_g3_r3","gold_accuracy":0.60,"measurement_only":true}

# 実装案/例: label-free 条件を壊さないアクセス制御
signals:
  proxy:
    visible_to: [solver, proposer, judge]
    sources:
      - execution_health
      - round_trip_consistency
      - public_tests
      - raw_execution_trace
  gold:
    visible_to: [measurement_pipeline_only]
    sources:
      - hidden_test_suite
      - gold_sql_result_set
      - heldout_rubric
    inject_into_harness: forbidden
    inject_into_trace: forbidden
    audit_log: required

# 実装案/例: 候補実行の timeout と fallback
def observe_with_guard(harness, batch, timeout_sec=90):
    try:
        return run_harness_in_subprocess(harness, batch, timeout=timeout_sec)
    except TimeoutError:
        return TraceResult(status="timeout", errors=["wall clock exceeded"])


def propose_with_fallback(parent, traces):
    child = run_proposer(parent, traces)
    if not child.loads_successfully():
        return parent
    if child.static_risk_score() > 0:
        return parent
    return child

# 実装案/例: proposer が生成する候補コードの実行制約
sandbox:
  filesystem:
    writable_paths:
      - workspace/harness_candidates
      - workspace/run_artifacts
    readonly_paths:
      - workspace/input_batches
    denied_paths:
      - workspace/gold
      - home_ssh
      - cloud_credentials
  network:
    egress: deny_all
  process:
    max_wall_clock_sec: 90
    max_memory_mb: 2048
  secrets:
    env_allowlist: []
    scrub_patterns: [API_KEY, TOKEN, PASSWORD, SECRET]

# 実装案/例: batch t で確定した harness を batch t+1 に適応前評価する
python -m tthe_eval.prequential \
  --stream data/unlabeled_stream.jsonl \
  --committed-harness-log runs/run_20260711/committed_harnesses.jsonl \
  --score-on next_batch_before_adaptation \
  --also-run shuffled_batch_control \
  --output runs/run_20260711/prequential_report.json

# 実装案/例: committed harness の昇格・巻き戻し条件
rollback_policy:
  promote_condition:
    min_holdout_delta: ">= 0.0"
    max_judge_gold_gap: "<= 0.20"
    max_timeout_rate: "<= 0.05"
  rollback_trigger:
    - hidden_score_drop_vs_last_good: "> 10pt"
    - judge_claimed_vs_gold_gap: "> 2/10"
    - timeout_rate: "> 5%"
  action_on_trigger: revert_to_last_known_good_harness
  keep_last_good_versions: 10

<!-- 実装案/例: judge 選定前チェックリスト -->
- [ ] 実行成功と意味的正しさを別々に確認しましたか
- [ ] 出力列数、粒度、型、単位を確認しましたか
- [ ] public tests だけで選んでいませんか
- [ ] raw trace を読み、エラー回復分岐の挙動を確認しましたか
- [ ] 最終 round の全候補を比較しましたか
- [ ] 候補間一致を正しさと誤認していませんか
- [ ] 追加 probe や再実行で judge の直感を検証しましたか