Swift/macOS 에이전트 앱에서 짧은 하드 타임아웃에 대한 CI 가드 에이전트 앱에서 짧은 하드 타임아웃은 CI에서 조용히 실패하는 원인이 되기 쉽습니다. 네트워크 호출, 프로세스 실행, 파일 시스템 작업에 고정된 짧은 타임아웃이 박혀 있으면, 로컬에서는 통과하던 테스트가 느린 CI 러너에서 간헐적으로 깨집니다. 이 글에서는 Swift/macOS 에이전트 앱에서 이런 짧은 하드 타임아웃이 코드베이스에 들어오는 것을 CI 단계에서 막는 방법을 다룹니다. 먼저 무엇을 "짧은 하드 타임아웃"으로 볼지 정의해야 합니다. 예를 들어 초 단위 리터럴로 직접 지정된 타임아웃, 설정에서 주입되지 않고 코드에 상수로 박힌 값, 그리고 테스트 환경에서 늘릴 수 없는 값을 대상으로 합니다. 반대로 설정 파일이나 환경 변수에서 읽어오는 타임아웃, 명시적으로 테스트용으로만 쓰이는 값은 예외로 둡니다. 가장 단순한 접근은 정적 검사입니다. Swift 소스에서 타임아웃 관련 API 호출을 찾고, 그 인자가 리터럴이거나 하드코딩된 상수인지 확인합니다. 예를 들어 `URLSession` 설정의 `timeoutIntervalForRequest`, `Process`의 대기 시간, `DispatchSemaphore.wait(timeout:)` 같은 호출을 패턴으로 잡습니다. 이 검사를 스크립트로 만들어 CI에 연결하면, 새로 추가되는 짧은 하드 타임아웃을 PR 단계에서 차단할 수 있습니다. 정적 검사만으로는 부족할 수 있습니다. 상수를 통해 간접적으로 전달되는 값이나 계산된 값은 패턴 매칭으로 잡기 어렵습니다. 이럴 때는 규칙을 조금 더 구조화합니다. 타임아웃 값을 반드시 중앙 설정 타입에서만 가져오도록 강제하고, 그 외 경로로 리터럴이 흘러들어오면 린트 규칙이나 커스텀 SwiftLint 규칙으로 잡습니다. 팀 규모가 크다면 이런 규칙을 문서화하고 CI에서 강제하는 편이 유지보수에 유리합니다. 런타임 가드도 함께 두면 좋습니다. 테스트 실행 시 환경 변수로 최소 타임아웃 하한을 설정하고, 앱 코드가 그 하한보다 짧은 값을 사용하려 하면 assertion이나 로그로 경고하게 합니다. CI에서는 이 하한을 넉넉하게 잡아 느린 러너에서도 간헐적 실패가 줄어듭니다. 다만 프로덕션 동작에는 영향을 주지 않도록 테스트 타깃이나 디버그 빌드에서만 활성화합니다. 마지막으로 CI 파이프라인에 이 검사들을 명시적으로 연결합니다. PR 검증 단계에서 정적 검사와 린트를 돌리고, 테스트 단계에서 런타임 하한을 적용합니다. 실패 시 어떤 파일과 라인이 문제인지 바로 알 수 있게 출력을 정리하면, 리뷰어와 작성자 모두 빠르게 고칠 수 있습니다. 이렇게 하면 짧은 하드 타임아웃이 조용히 쌓이는 것을 막고, CI 안정성을 유지할 수 있습니다.

# Short Timeout CI Guard

I'll add a shrink-only allowlist guard that catches new hard timeout literals below 30 seconds across the app/daemon, CLI, and agent-session paths.

## 1. `scripts/ci/check_short_timeouts.py`

```python
#!/usr/bin/env python3
"""Fail CI when new hard timeout literals below 30 seconds are introduced.

Scans the app/daemon, CLI, and agent-session paths for hard-coded timeout
literals shorter than 30 seconds. Existing violations are captured in
`tests/short-timeout-allowlist.json` as a shrink-only allowlist: entries may
be removed but never added without an adjacent `// timeout: <reason>` or
`# timeout: <reason>` comment justifying the bound.

Usage:
    python scripts/ci/check_short_timeouts.py            # check
    python scripts/ci/check_short_timeouts.py --update   # regenerate snapshot
"""

from __future__ import annotations

import argparse
import json
import re
import sys
from dataclasses import dataclass
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parents[2]
ALLOWLIST_PATH = REPO_ROOT / "tests" / "short-timeout-allowlist.json"

# Directories scanned for short timeout literals.
SCAN_ROOTS = [
    "app",
    "daemon",
    "cli",
    "agent-session",
    "agent_session",
]

# File extensions we understand.
SCAN_SUFFIXES = {
    ".rs",
    ".ts",
    ".tsx",
    ".js",
    ".jsx",
    ".py",
    ".swift",
    ".go",
}

MIN_TIMEOUT_SECONDS = 30.0

# A `// timeout: <reason>` or `# timeout: <reason>` comment on the same line
# (or the line immediately above) justifies a short bound.
REASON_RE = re.compile(r"(?://|#)\s*timeout:\s*(?P<reason>\S.*)")

# Rust test modules and files are excluded.
RUST_TEST_FILE_RE = re.compile(r"_tests\.rs$")
RUST_CFG_TEST_RE = re.compile(r"#\[cfg\(test\)\]")

# ---------------------------------------------------------------------------
# Timeout literal patterns
# ---------------------------------------------------------------------------
#
# Each pattern yields (value_in_seconds, unit) so we can normalize to seconds.
# Patterns are intentionally conservative: they match named timeout/deadline
# identifiers followed by a numeric literal and a unit.

# Named identifiers that imply a timeout/deadline.
_TIMEOUT_IDENT = r"(?:timeout|deadline|time_limit|timeLimit|max_wait|maxWait|wait_timeout|waitTimeout|connect_timeout|connectTimeout|read_timeout|readTimeout|idle_timeout|idleTimeout|ttl|TTL)"

# Milliseconds: `timeout_ms = 500`, `timeoutMs: 500`, `timeout_ms(500)`.
MS_PATTERN = re.compile(
    rf"(?P<ident>{_TIMEOUT_IDENT})[_A-Za-z]*\s*[:=(]\s*(?P<value>\d+(?:\.\d+)?)\s*(?:ms|millis|milliseconds)?\b",
    re.IGNORECASE,
)

# Seconds: `timeout = 5`, `timeout_secs: 5`, `timeoutSeconds(5)`.
SEC_PATTERN = re.compile(
    rf"(?P<ident>{_TIMEOUT_IDENT})[_A-Za-z]*\s*[:=(]\s*(?P<value>\d+(?:\.\d+)?)\s*(?:s|secs|seconds)?\b",
    re.IGNORECASE,
)

# Rust Duration: `Duration::from_secs(5)`, `Duration::from_millis(500)`.
RUST_DURATION_RE = re.compile(
    r"Duration::from_(?P<unit>secs|millis|micros)\s*\(\s*(?P<value>\d+(?:\.\d+)?)\s*\)"
)

# Swift TimeInterval / DispatchTimeInterval: `.seconds(5)`, `.milliseconds(500)`.
SWIFT_DURATION_RE = re.compile(
    r"\.(?P<unit>seconds|milliseconds|microseconds)\s*\(\s*(?P<value>\d+(?:\.\d+)?)\s*\)"
)

# Go time.Duration: `5 * time.Second`, `500 * time.Millisecond`.
GO_DURATION_RE = re.compile(
    r"(?P<value>\d+(?:\.\d+)?)\s*\*\s*time\.(?P<unit>Second|Millisecond|Microsecond)"
)

# Python: `timeout=5`, `timeout=0.5`.
PY_TIMEOUT_RE = re.compile(
    r"\btimeout\s*=\s*(?P<value>\d+(?:\.\d+)?)\b"
)


@dataclass(frozen=True)
class Violation:
    path: str
    line: int
    value_seconds: float
    reason: str | None

    @property
    def key(self) -> str:
        return f"{self.path}:{self.line}"

    def to_json(self) -> dict:
        entry = {"path": self.path, "line": self.line, "seconds": self.value_seconds}
        if self.reason:
            entry["reason"] = self.reason
        return entry


def _normalize(unit: str, value: float) -> float:
    unit = unit.lower()
    if unit in ("ms", "millis", "milliseconds", "millisecond"):
        return value / 1000.0
    if unit in ("micros", "microseconds", "microsecond"):
        return value / 1_000_000.0
    return value


def _iter_source_files() -> list[Path]:
    files: list[Path] = []
    for root_name in SCAN_ROOTS:
        root = REPO_ROOT / root_name
        if not root.exists():
            continue
        for path in root.rglob("*"):
            if not path.is_file():
                continue
            if path.suffix not in SCAN_SUFFIXES:
                continue
            if path.suffix == ".rs" and RUST_TEST_FILE_RE.search(path.name):
                continue
            files.append(path)
    return sorted(files)


def _rust_test_module_lines(lines: list[str]) -> set[int]:
    """Return 1-based line numbers inside `#[cfg(test)]` modules."""
    excluded: set[int] = set()
    depth = 0
    in_test_mod = False
    for idx, line in enumerate(lines, start=1):
        if RUST_CFG_TEST_RE.search(line):
            in_test_mod = True
        if in_test_mod:
            excluded.add(idx)
            depth += line.count("{") - line.count("}")
            if depth <= 0 and "{" in "".join(lines[:idx]):
                # Module closed; stop excluding once braces balance.
                if depth == 0 and idx > 1:
                    in_test_mod = False
    return excluded


def _find_reason(lines: list[str], idx: int) -> str | None:
    """Look for a `timeout: <reason>` comment on this line or the one above."""
    for candidate in (lines[idx], lines[idx - 1] if idx > 0 else ""):
        match = REASON_RE.search(candidate)
        if match:
            return match.group("reason").strip()
    return None


def _scan_line(line: str) -> list[tuple[float, str]]:
    """Return (seconds, unit) tuples for timeout literals on a line."""
    hits: list[tuple[float, str]] = []

    for match in RUST_DURATION_RE.finditer(line):
        hits.append((_normalize(match.group("unit"), float(match.group("value"))), match.group("unit")))

    for match in SWIFT_DURATION_RE.finditer(line):
        hits.append((_normalize(match.group("unit"), float(match.group("value"))), match.group("unit")))

    for match in GO_DURATION_RE.finditer(line):
        hits.append((_normalize(match.group("unit"), float(match.group("value"))), match.group("unit")))

    for match in PY_TIMEOUT_RE.finditer(line):
        hits.append((float(match.group("value")), "s"))

    for pattern, default_unit in ((MS_PATTERN, "ms"), (SEC_PATTERN, "s")):
        for match in pattern.finditer(line):
            ident = match.group("ident").lower()
            # Skip identifiers that already carry an explicit unit suffix.
            if ident.endswith(("ms", "millis", "milliseconds")):
                unit = "ms"
            elif ident.endswith(("secs", "seconds")):
                unit = "s"
            else:
                unit = default_unit
            hits.append((_normalize(unit, float(match.group("value"))), unit))

    return hits


def scan() -> list[Violation]:
    violations: list[Violation] = []
    for path in _iter_source_files():
        try:
            text = path.read_text(encoding="utf-8")
        except (UnicodeDecodeError, OSError):
            continue
        lines = text.splitlines()
        excluded = _rust_test_module_lines(lines) if path.suffix == ".rs" else set()
        rel = str(path.relative_to(REPO_ROOT))
        for idx, line in enumerate(lines):
            lineno = idx + 1
            if lineno in excluded:
                continue
            for seconds, _unit in _scan_line(line):
                if seconds >= MIN_TIMEOUT_SECONDS:
                    continue
                reason = _find_reason(lines, idx)
                violations.append(
                    Violation(path=rel, line=lineno, value_seconds=seconds, reason=reason)
                )
    return violations


def load_allowlist() -> dict[str, dict]:
    if not ALLOWLIST_PATH.exists():
        return {}
    data = json.loads(ALLOWLIST_PATH.read_text(encoding="utf-8"))
    return {entry["key"]: entry for entry in data.get("entries", [])}


def write_allowlist(violations: list[Violation], previous: dict[str, dict]) -> None:
    entries = []
    for v in sorted(violations, key=lambda x: (x.path, x.line)):
        entry = v.to_json()
        # Preserve a previously recorded reason if the current scan lacks one.
        if "reason" not in entry and v.key in previous and "reason" in previous[v.key]:
            entry["reason"] = previous[v.key]["reason"]
        entries.append(entry)
    payload = {
        "min_timeout_seconds": MIN_TIMEOUT_SECONDS,
        "entries": entries,
    }
    ALLOWLIST_PATH.parent.mkdir(parents=True, exist_ok=True)
    ALLOWLIST_PATH.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")


def check() -> int:
    violations = scan()
    allowlist = load_allowlist()

    new_violations = [v for v in violations if v.key not in allowlist]
    # A violation with a reason is always allowed, even if not in the allowlist.
    unjustified = [v for v in new_violations if not v.reason]

    # Ratchet: allowlist entries that no longer correspond to a violation must
    # be removed (shrink-only). We report them so `--update` can prune.
    current_keys = {v.key for v in violations}
    stale = [key for key in allowlist if key not in current_keys]

    failed = False
    if unjustified:
        failed = True
        print("New short timeout literals below 30s without a `timeout:` reason:", file=sys.stderr)
        for v in unjustified:
            print(f"  {v.path}:{v.line}  ({v.value_seconds:g}s)", file=sys.stderr)
        print(
            "\nAdd an adjacent `// timeout: <reason>` or `# timeout: <reason>` comment, "
            "or run `python scripts/ci/check_short_timeouts.py --update` to snapshot "
            "existing violations.",
            file=sys.stderr,
        )

    if stale:
        failed = True
        print("Allowlist entries no longer match any violation (shrink-only):", file=sys.stderr)
        for key in stale:
            print(f"  {key}", file=sys.stderr)
        print(
            "\nRun `python scripts/ci/check_short_timeouts.py --update` to prune.",
            file=sys.stderr,
        )

    if not failed:
        print(f"OK: {len(violations)} allowlisted short timeout(s), no new violations.")
    return 1 if failed else 0


def update() -> int:
    previous = load_allowlist()
    violations = scan()
    write_allowlist(violations, previous)
    print(f"Updated {ALLOWLIST_PATH.relative_to(REPO_ROOT)} with {len(violations)} entries.")
    return 0


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--update",
        action="store_true",
        help="Regenerate the allowlist snapshot, preserving existing reasons.",
    )
    args = parser.parse_args(argv)
    return update() if args.update else check()


if __name__ == "__main__":
    raise SystemExit(main())
```

## 2. `tests/short-timeout-allowlist.json`

```json
{
  "min_timeout_seconds": 30.0,
  "entries": []
}
```

## 3. `tests/test_short_timeout_guard.py`

```python
"""Tests for scripts/ci/check_short_timeouts.py."""

from __future__ import annotations

import importlib.util
import json
import sys
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parents[1]
SCRIPT_PATH = REPO_ROOT / "scripts" / "ci" / "check_short_timeouts.py"


def _load_module():
    spec = importlib.util.spec_from_file_location("check_short_timeouts", SCRIPT_PATH)
    module = importlib.util.module_from_spec(spec)
    sys.modules["check_short_timeouts"] = module
    spec.loader.exec_module(module)
    return module


@pytest.fixture()
def guard(tmp_path, monkeypatch):
    module = _load_module()
    # Redirect the module's repo root and allowlist to a temp sandbox.
    monkeypatch.setattr(module, "REPO_ROOT", tmp_path)
    monkeypatch.setattr(module, "ALLOWLIST_PATH", tmp_path / "tests" / "short-timeout-allowlist.json")
    (tmp_path / "tests").mkdir()
    return module


def _write(root: Path, rel: str, content: str) -> Path:
    path = root / rel
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content, encoding="utf-8")
    return path


def test_new_two_second_deadline_fails(guard, tmp_path):
    _write(tmp_path, "app/client.ts", "const deadline = 2;\n")
    assert guard.check() == 1


def test_reason_comment_passes(guard, tmp_path):
    _write(tmp_path, "app/client.ts", "const deadline = 2; // timeout: hung peer\n")
    assert guard.check() == 0


def test_hash_reason_comment_passes(guard, tmp_path):
    _write(tmp_path, "cli/run.py", "timeout = 5  # timeout: legacy peer\n")
    assert guard.check() == 0


def test_wrapped_swift_duration(guard, tmp_path):
    _write(
        tmp_path,
        "app/Client.swift",
        "let t = DispatchTimeInterval\n    .seconds(5)\n",
    )
    assert guard.check() == 1


def test_millisecond_units(guard, tmp_path):
    _write(tmp_path, "app/client.ts", "const timeoutMs = 500;\n")
    assert guard.check() == 1


def test_rust_duration_millis(guard, tmp_path):
    _write(tmp_path, "daemon/net.rs", "let t = Duration::from_millis(250);\n")
    assert guard.check() == 1


def test_go_duration(guard, tmp_path):
    _write(tmp_path, "daemon/net.go", "t := 5 * time.Second\n")
    assert guard.check() == 1


def test_python_timeout_kwarg(guard, tmp_path):
    _write(tmp_path, "cli/http.py", "client.get(url, timeout=3)\n")
    assert guard.check() == 1


def test_rust_test_file_excluded(guard, tmp_path):
    _write(tmp_path, "daemon/net_tests.rs", "let t = Duration::from_secs(1);\n")
    assert guard.check() == 0


def test_rust_cfg_test_module_excluded(guard, tmp_path):
    _write(
        tmp_path,
        "daemon/net.rs",
        "fn real() {}\n\n#[cfg(test)]\nmod tests {\n    fn t() { let _ = Duration::from_secs(1); }\n}\n",
    )
    assert guard.check() == 0


def test_ratchet_growth_fails(guard, tmp_path):
    _write(tmp_path, "app/a.ts", "const timeout = 5;\n")
    assert guard.update() == 0
    _write(tmp_path, "app/b.ts", "const timeout = 5;\n")
    assert guard.check() == 1


def test_ratchet_shrink_requires_update(guard, tmp_path):
    _write(tmp_path, "app/a.ts", "const timeout = 5;\n")
    assert guard.update() == 0
    (tmp_path / "app" / "a.ts").unlink()
    # Stale entry must be pruned via --update.
    assert guard.check() == 1
    assert guard.update() == 0
    assert guard.check() == 0


def test_update_preserves_reasons(guard, tmp_path):
    _write(tmp_path, "app/a.ts", "const timeout = 5; // timeout: hung peer\n")
    assert guard.update() == 0
    data = json.loads(guard.ALLOWLIST_PATH.read_text())
    assert data["entries"][0]["reason"] == "hung peer"
```

## 4. `tests/test-execution.toml`

```toml
[[test]]
name = "short-timeout-guard"
command = "python scripts/ci/check_short_timeouts.py"
tags = ["linux-guard"]
```

## 5. `.github/workflows/ci-guards.yml`

```yaml
name: CI Guards

on:
  pull_request:
  push:
    branches: [main]

jobs:
  short-timeout-guard:
    name: short-timeout-guard
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - name: Run short timeout guard
        run: python scripts/ci/check_short_timeouts.py
      - name: Run guard unit tests
        run: python -m pytest tests/test_short_timeout_guard.py -q
```

## Notes

- **Shrink-only ratchet**: `check()` fails on both new unjustified violations *and* stale allowlist entries, forcing `--update` to prune. `--update` preserves any `reason` already recorded for a matching `path:line`.
- **Reason escape hatch**: an adjacent `// timeout: <reason>` or `# timeout: <reason>` comment (same line or line above) always permits a short bound, even if not in the allowlist.
- **Rust exclusions**: `*_tests.rs` files are skipped entirely; `#[cfg(test)]` module bodies are excluded by brace tracking.
- **Unit normalization**: milliseconds, microseconds, seconds, and Swift/Go/Rust duration constructors all normalize to seconds before the 30s comparison.