awoooi/docs/superpowers/plans/2026-04-19-aider-watch.md

aider-watch Implementation Plan（v1 — ⛔ DEPRECATED 2026-04-20）

⛔ 已被 v2 取代 — 統帥於 2026-04-20 改走「完全整合進 awoooi」路線。 新版 plan：2026-04-20-aider-watch-v2.md v1 Task 0-5 已完成（~/aider-watch 8 commits），events.py + redactor.py 會搬進 awoooi。

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: 建立 aider CLI 的全程監控系統：Python wrapper 攔截 aider stdout + chat history → Telegram DM 即時推播 + PG 188 儲存 + 每日/週報 launchd 排程。

Architecture: Hybrid Python wrapper (stdout 即時訊號 + chat_history.md canonical events) → dispatcher → Telegram/PG/JSONL 三分流。PG 不可達時 fallback JSONL buffer，launchd 每 5min flush 補寫。launchd 另有 daily/weekly reporter job。

Tech Stack: Python 3.11.7 (pyenv)、pipx、pydantic v2、psycopg2-binary、requests、pexpect、pytest、PostgreSQL 192.168.0.188:5432、launchd

Spec: docs/superpowers/specs/2026-04-19-aider-watch-design.md

---

File Structure

New git repo at ~/aider-watch/（獨立於 awoooi）：

檔案	職責
pyproject.toml	套件定義、依賴、CLI entrypoints
bin/aiderw	Python shim，轉呼 aider_watch.wrapper:main
aider_watch/__init__.py	套件 metadata
aider_watch/events.py	Pydantic v2 event dataclasses + JSON serde
aider_watch/redactor.py	Secret pattern 遮罩（OpenRouter/GitHub/Google/OpenAI/Anthropic/AWS）
aider_watch/telegram.py	Bot API client（format + rate limit + 429 backoff）
aider_watch/storage.py	PG writer + JSONL fallback + flush
aider_watch/parsers.py	aider stdout 解析 + chat_history.md 重建
aider_watch/wrapper.py	主入口：subprocess + dispatcher + silent timeout + atexit
aider_watch/reporter.py	日/週報聚合查詢 + TG 格式
aider_watch/cli.py	aider-watch {doctor,flush,replay,report} 派發
aider_watch/config.py	讀 ~/.aider-watch.env、路徑常數
schema.sql	PG DDL（sessions + events + file_touches）
scripts/install_pg.sh	PG 188 上建 DB/user + 套 schema（one-off）
scripts/install_launchd.sh	安裝 3 支 plist + 啟用
scripts/install.sh	總編：pipx install + .env 產生 + symlink + DB + launchd
launchd/com.awoooi.aider-watch.flush.plist	每 5min
launchd/com.awoooi.aider-watch.daily.plist	每日 23:50
launchd/com.awoooi.aider-watch.weekly.plist	週日 22:00
tests/test_events.py	Unit: event schema round-trip
tests/test_redactor.py	Unit: secret 遮罩
tests/test_telegram_format.py	Unit: event → markdown
tests/test_parsers.py	Unit: aider stdout + chat history 解析
tests/test_storage_pg.py	Integration: PG round-trip（用 aider_watch_test DB）
tests/test_storage_buffer.py	Integration: JSONL fallback + flush
tests/test_telegram_live.py	Integration: 真發 [TEST] 訊息
tests/test_e2e_happy.py	E2E: aiderw --message ... --exit
tests/test_e2e_degradation.py	E2E: 斷網 → buffer → flush
tests/test_e2e_report.py	E2E: daily/weekly report

不在 awoooi repo 內：aider-watch 是統帥的個人工具，獨立 git repo 易 clone 到新機器。

---

Task 0: 建立專案骨架

Files:

• Create: ~/aider-watch/pyproject.toml

• Create: ~/aider-watch/aider_watch/__init__.py

• Create: ~/aider-watch/.gitignore

• Create: ~/aider-watch/README.md

• Step 1: 建目錄、git init

mkdir -p ~/aider-watch/{aider_watch,tests,scripts,launchd,bin}
cd ~/aider-watch
git init -b main

• Step 2: 寫 pyproject.toml

[project]
name = "aider-watch"
version = "0.1.0"
description = "aider CLI 監控：即時 Telegram 推播 + PG 儲存 + 日週報"
requires-python = ">=3.11"
dependencies = [
    "pydantic>=2.6,<3",
    "psycopg2-binary>=2.9",
    "requests>=2.31",
    "pexpect>=4.9",
    "python-ulid>=3.0",
    "python-dateutil>=2.9",
]

[project.optional-dependencies]
dev = ["pytest>=8.0", "pytest-cov>=5.0"]

[project.scripts]
aiderw = "aider_watch.wrapper:main"
aider-watch = "aider_watch.cli:main"

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

[tool.setuptools.packages.find]
include = ["aider_watch*"]

• Step 3: 寫 .gitignore

__pycache__/
*.pyc
.pytest_cache/
.coverage
htmlcov/
dist/
build/
*.egg-info/
.env
.aider-watch.env
.DS_Store

• Step 4: 寫 aider_watch/init.py

# aider-watch — 統帥本機 aider CLI 監控 | 2026-04-19 建立 @ Asia/Taipei
__version__ = "0.1.0"

• Step 5: 寫 README.md 簡要

# aider-watch

統帥本機 aider CLI 全程監控（Telegram + PG + 日週報）。

參考設計稿：awoooi/docs/superpowers/specs/2026-04-19-aider-watch-design.md

• Step 6: Commit

cd ~/aider-watch
git add -A
git commit -m "chore: scaffold aider-watch project"

---

Task 1: Event Schema（pydantic v2 dataclasses）

Files:

• Create: ~/aider-watch/aider_watch/events.py

• Create: ~/aider-watch/tests/test_events.py

• Create: ~/aider-watch/tests/__init__.py

• Step 1: 寫測試 tests/test_events.py

# 2026-04-19 @ Asia/Taipei
from datetime import datetime, timezone, timedelta
from aider_watch.events import (
    Event, SessionStart, FileEdit, ErrorEvent, Commit, SilentTimeout, SessionEnd
)
import json

TAIPEI = timezone(timedelta(hours=8))

def test_session_start_roundtrip():
    ev = SessionStart(
        ts=datetime(2026, 4, 19, 22, 0, tzinfo=TAIPEI),
        session_id="01J7XZABC",
        payload={"cwd": "/tmp/x", "model": "elephant-alpha",
                 "aider_args": ["-m", "hi"], "aider_pid": 123, "cli_version": "0.86.2"},
    )
    j = ev.model_dump_json()
    ev2 = Event.from_json(j)
    assert ev2.type == "session_start"
    assert ev2.payload["model"] == "elephant-alpha"
    assert ev2.ts.tzinfo is not None  # timezone-aware

def test_file_edit_roundtrip():
    ev = FileEdit(
        ts=datetime.now(TAIPEI),
        session_id="01J7XZABC",
        payload={"path": "apps/api/x.py", "lines_added": 12,
                 "lines_deleted": 3, "diff_head": ["@@ def f():", "+    pass", "-    old"],
                 "is_new_file": False},
    )
    assert ev.type == "file_edit"
    data = json.loads(ev.model_dump_json())
    assert data["payload"]["lines_added"] == 12

def test_session_end_fields():
    ev = SessionEnd(
        ts=datetime.now(TAIPEI),
        session_id="s1",
        payload={"duration_sec": 332, "tokens_sent": 12400, "tokens_received": 3800,
                 "cost_usd": 0, "files_changed": 3, "error_count": 0, "exit_code": 0},
    )
    assert ev.payload["duration_sec"] == 332

def test_event_type_immutable():
    ev = SessionStart(ts=datetime.now(TAIPEI), session_id="s",
                      payload={"cwd": "/t", "model": "m", "aider_args": [],
                               "aider_pid": 0, "cli_version": "x"})
    assert ev.type == "session_start"

• Step 2: 跑測試驗失敗

cd ~/aider-watch && pip install -e '.[dev]' --quiet
pytest tests/test_events.py -v

Expected: ModuleNotFoundError: No module named 'aider_watch.events'

• Step 3: 寫實作 aider_watch/events.py

# aider-watch event schema | 2026-04-19 @ Asia/Taipei
"""
Event 資料契約：7 種 event type，統一外殼 + type-specific payload。
所有 timestamp 必須 timezone-aware（台北 +08:00）。
"""
from __future__ import annotations
from datetime import datetime
from typing import Any, Literal, Union
from pydantic import BaseModel, Field, ConfigDict
import json


class _EventBase(BaseModel):
    model_config = ConfigDict(frozen=False)
    ts: datetime
    session_id: str
    payload: dict[str, Any]

    def model_dump_json(self, **kw) -> str:  # type: ignore[override]
        return json.dumps(
            {"ts": self.ts.isoformat(), "session_id": self.session_id,
             "type": self.type, "payload": self.payload},
            ensure_ascii=False, default=str,
        )


class SessionStart(_EventBase):
    type: Literal["session_start"] = "session_start"


class FileEdit(_EventBase):
    type: Literal["file_edit"] = "file_edit"


class ErrorEvent(_EventBase):
    type: Literal["error"] = "error"


class Commit(_EventBase):
    type: Literal["commit"] = "commit"


class SilentTimeout(_EventBase):
    type: Literal["silent_timeout"] = "silent_timeout"


class SessionEnd(_EventBase):
    type: Literal["session_end"] = "session_end"


class Raw(_EventBase):
    type: Literal["raw"] = "raw"


Event = Union[SessionStart, FileEdit, ErrorEvent, Commit, SilentTimeout, SessionEnd, Raw]

_TYPE_MAP = {
    "session_start": SessionStart, "file_edit": FileEdit, "error": ErrorEvent,
    "commit": Commit, "silent_timeout": SilentTimeout,
    "session_end": SessionEnd, "raw": Raw,
}


def from_json(s: str) -> _EventBase:
    d = json.loads(s)
    cls = _TYPE_MAP[d["type"]]
    return cls(ts=datetime.fromisoformat(d["ts"]),
               session_id=d["session_id"], payload=d["payload"])


# attach classmethod for convenience
_EventBase.from_json = staticmethod(from_json)  # type: ignore[attr-defined]

• Step 4: 跑測試驗通過

pytest tests/test_events.py -v

Expected: 4 passed

• Step 5: Commit

git add aider_watch/events.py tests/test_events.py tests/__init__.py
git commit -m "feat(events): pydantic v2 event schema with 7 types"

---

Task 2: Secret Redactor

Files:

• Create: ~/aider-watch/aider_watch/redactor.py

• Create: ~/aider-watch/tests/test_redactor.py

• Step 1: 寫測試 tests/test_redactor.py

# 2026-04-19 @ Asia/Taipei
from aider_watch.redactor import redact

def test_openrouter_key_redacted():
    s = "failed with sk-or-v1-8ad9d715327496e71e30d1e50cc00903a1ece23f0"
    assert "sk-or-v1-8ad9" not in redact(s)
    assert "<redacted:openrouter>" in redact(s)

def test_github_token_redacted():
    assert "<redacted:github>" in redact("ghp_abcdef0123456789ABCDEFghijklmnopqrst")

def test_google_api_key_redacted():
    assert "<redacted:google>" in redact("AIzaSyABCDEFGHIJKLMNOPQRSTUVWXYZ1234567")

def test_openai_key_redacted():
    assert "<redacted:openai>" in redact("sk-abcdEFGH1234567890abcdEFGH1234567890abcdEFGH12")

def test_anthropic_key_redacted():
    assert "<redacted:anthropic>" in redact("sk-ant-api03-abcDEF_123-xyz")

def test_telegram_bot_token_redacted():
    assert "<redacted:telegram>" in redact("8474499448:AAFqu_i4-PN4zGFOK5ea8o0Ud56qqEtCMeI")

def test_clean_text_passthrough():
    s = "hello world, nothing secret here"
    assert redact(s) == s

def test_dict_recursive():
    d = {"msg": "token=ghp_abcdef0123456789ABCDEFghijklmnopqrst",
         "nested": {"tg": "8474499448:AAFqu_i4-PN4zGFOK5ea8o0Ud56qqEtCMeI"}}
    out = redact(d)
    assert "ghp_abcdef" not in str(out)
    assert "<redacted:github>" in out["msg"]
    assert "<redacted:telegram>" in out["nested"]["tg"]

• Step 2: 跑測試驗失敗

pytest tests/test_redactor.py -v

Expected: ModuleNotFoundError: No module named 'aider_watch.redactor'

• Step 3: 寫實作 aider_watch/redactor.py

# aider-watch secret redactor | 2026-04-19 @ Asia/Taipei
"""把文字/dict 中的 secret 遮罩為 <redacted:kind>。符合 feedback_secrets_leak_incidents_2026-04-18.md。"""
from __future__ import annotations
import re
from typing import Any

_PATTERNS = [
    (re.compile(r"sk-or-v1-[A-Za-z0-9]{40,}"),        "openrouter"),
    (re.compile(r"sk-ant-api\d{2}-[A-Za-z0-9_\-]{20,}"), "anthropic"),
    (re.compile(r"sk-[A-Za-z0-9]{40,}"),              "openai"),       # 注意順序：排在 openrouter/anthropic 之後
    (re.compile(r"ghp_[A-Za-z0-9]{36}"),              "github"),
    (re.compile(r"AIza[0-9A-Za-z_\-]{35}"),           "google"),
    (re.compile(r"\b\d{8,10}:[A-Za-z0-9_\-]{35}\b"),  "telegram"),
    (re.compile(r"AKIA[0-9A-Z]{16}"),                 "aws"),
]


def redact(obj: Any) -> Any:
    """遮罩字串/dict/list 內 secret，不影響其他型別。"""
    if isinstance(obj, str):
        s = obj
        for pat, kind in _PATTERNS:
            s = pat.sub(f"<redacted:{kind}>", s)
        return s
    if isinstance(obj, dict):
        return {k: redact(v) for k, v in obj.items()}
    if isinstance(obj, list):
        return [redact(x) for x in obj]
    return obj

• Step 4: 跑測試驗通過

pytest tests/test_redactor.py -v

Expected: 8 passed

• Step 5: Commit

git add aider_watch/redactor.py tests/test_redactor.py
git commit -m "feat(redactor): 7 secret patterns (openrouter/anthropic/openai/gh/google/tg/aws)"

---

Task 3: Config Loader

Files:

• Create: ~/aider-watch/aider_watch/config.py

• Step 1: 寫 aider_watch/config.py

# aider-watch config | 2026-04-19 @ Asia/Taipei
"""讀 ~/.aider-watch.env 為環境變數 + 集中暴露設定常數。"""
from __future__ import annotations
import os
from pathlib import Path
from datetime import timezone, timedelta

HOME = Path.home()
WATCH_ROOT = HOME / "aider-watch"
SESSIONS_DIR = WATCH_ROOT / "sessions"
BUFFER_DIR = WATCH_ROOT / "buffer"
LIVE_LOG = WATCH_ROOT / "live.log"
STATE_FILE = WATCH_ROOT / "state.json"
LOGS_DIR = WATCH_ROOT / "logs"
ENV_FILE = HOME / ".aider-watch.env"

TAIPEI = timezone(timedelta(hours=8))


def load_env() -> None:
    """把 ~/.aider-watch.env 的 KEY=VALUE 行寫入 os.environ（不覆寫已存在的）。"""
    if not ENV_FILE.exists():
        return
    for line in ENV_FILE.read_text().splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, _, v = line.partition("=")
        k, v = k.strip(), v.strip()
        os.environ.setdefault(k, v)


def get(name: str, default: str | None = None, required: bool = False) -> str:
    load_env()
    v = os.environ.get(name, default)
    if required and not v:
        raise RuntimeError(f"{name} 未設定（check {ENV_FILE}）")
    return v or ""


def ensure_dirs() -> None:
    for d in (WATCH_ROOT, SESSIONS_DIR, BUFFER_DIR, LOGS_DIR):
        d.mkdir(parents=True, exist_ok=True)

• Step 2: Commit（無測試因無邏輯；Task 4+ 間接測到）

git add aider_watch/config.py
git commit -m "feat(config): env loader + path constants"

---

Task 4: Telegram Formatter（純函式，先測 format 再測 send）

Files:

• Create: ~/aider-watch/aider_watch/telegram.py（format 部分）

• Create: ~/aider-watch/tests/test_telegram_format.py

• Step 1: 寫測試 tests/test_telegram_format.py

# 2026-04-19 @ Asia/Taipei
from datetime import datetime
from aider_watch.events import SessionStart, FileEdit, ErrorEvent, Commit, SilentTimeout, SessionEnd
from aider_watch.telegram import format_event
from aider_watch.config import TAIPEI

def _ts(): return datetime(2026, 4, 19, 22, 0, tzinfo=TAIPEI)

def test_format_session_start():
    ev = SessionStart(ts=_ts(), session_id="s1",
                      payload={"cwd": "/Users/ogt/awoooi", "model": "elephant-alpha",
                               "aider_args": [], "aider_pid": 12345, "cli_version": "0.86.2"})
    m = format_event(ev)
    assert "🚀" in m
    assert "/Users/ogt/awoooi" in m
    assert "elephant-alpha" in m

def test_format_file_edit_with_diff():
    ev = FileEdit(ts=_ts(), session_id="s1",
                  payload={"path": "apps/api/foo.py", "lines_added": 12,
                           "lines_deleted": 3,
                           "diff_head": ["@@ def f():", "+    log()", "-    pass"],
                           "is_new_file": False})
    m = format_event(ev)
    assert "✏️" in m and "foo.py" in m and "+12" in m and "-3" in m
    assert "log()" in m

def test_format_session_end():
    ev = SessionEnd(ts=_ts(), session_id="s1",
                    payload={"duration_sec": 332, "tokens_sent": 12400,
                             "tokens_received": 3800, "cost_usd": 0,
                             "files_changed": 3, "error_count": 0, "exit_code": 0})
    m = format_event(ev)
    assert "🏁" in m and "5m32s" in m and "12.4k" in m and "exit=0" in m

def test_format_redacts_secrets():
    ev = ErrorEvent(ts=_ts(), session_id="s1",
                    payload={"kind": "api_auth", "message": "bad key sk-or-v1-abcdef0123456789ABCDEFghijklmnopqrstuv",
                             "context_50chars": ""})
    m = format_event(ev)
    assert "sk-or-v1-abcdef" not in m
    assert "<redacted:" in m

• Step 2: 跑測試驗失敗

pytest tests/test_telegram_format.py -v

Expected: ModuleNotFoundError: No module named 'aider_watch.telegram'

• Step 3: 寫 aider_watch/telegram.py（format 部分）

# aider-watch telegram | 2026-04-19 @ Asia/Taipei
"""Telegram Bot API client：format + send with backoff + fail-open（不阻塞 aider）。"""
from __future__ import annotations
import time
from typing import Any
import requests
from aider_watch.events import (
    _EventBase, SessionStart, FileEdit, ErrorEvent, Commit, SilentTimeout, SessionEnd
)
from aider_watch.redactor import redact
from aider_watch.config import get


def _fmt_dur(sec: int) -> str:
    m, s = divmod(max(sec, 0), 60)
    if m >= 60:
        h, m = divmod(m, 60)
        return f"{h}h{m}m{s}s"
    return f"{m}m{s}s"


def _fmt_tokens(n: int) -> str:
    if n >= 1000:
        return f"{n/1000:.1f}k"
    return str(n)


def format_event(ev: _EventBase) -> str | None:
    """Event → Telegram markdown；不該推到 TG 的（raw）回 None。"""
    p = redact(ev.payload)
    t = ev.type
    if t == "session_start":
        return (f"🚀 <b>aider 啟動</b>\n"
                f"repo: <code>{p['cwd']}</code>\n"
                f"model: {p['model']}\n"
                f"pid: {p['aider_pid']}")
    if t == "file_edit":
        head = "\n".join(f"    {l}" for l in p.get("diff_head", [])[:3])
        tag = "🆕" if p.get("is_new_file") else "✏️"
        return (f"{tag} <b>edit</b> <code>{p['path']}</code> "
                f"(+{p['lines_added']} -{p['lines_deleted']})\n{head}")
    if t == "error":
        return f"❌ <b>error</b>: {p['kind']} — {p['message'][:200]}"
    if t == "commit":
        files = ", ".join(p.get("files", [])[:5])
        return f"📌 <b>commit</b> <code>{p['sha'][:7]}</code>: {p['message']}\n→ {files}"
    if t == "silent_timeout":
        return f"⏸️ <b>aider 靜默</b> {p['idle_sec']//60} 分鐘 (last: <i>{p.get('last_output_tail','')[:40]}</i>)"
    if t == "session_end":
        dur = _fmt_dur(p["duration_sec"])
        return (f"🏁 <b>session 結束</b>\n"
                f"⏱️ {dur} | 🎯 exit={p['exit_code']}\n"
                f"📊 {_fmt_tokens(p['tokens_sent'])}↑ / {_fmt_tokens(p['tokens_received'])}↓ tokens | ${p['cost_usd']}\n"
                f"📝 改 {p['files_changed']} 檔 | ❌ {p['error_count']} error")
    return None  # raw 不推

• Step 4: 跑測試驗通過

pytest tests/test_telegram_format.py -v

Expected: 4 passed

• Step 5: Commit

git add aider_watch/telegram.py tests/test_telegram_format.py
git commit -m "feat(telegram): format_event for 6 event types + redact"

---

Task 5: Telegram Send（backoff + fail-open）

Files:

• Modify: ~/aider-watch/aider_watch/telegram.py（補 send）

• Create: ~/aider-watch/tests/test_telegram_live.py

• Step 1: 補實作 send() 到 aider_watch/telegram.py 尾端

# === 續 telegram.py ===

_API = "https://api.telegram.org/bot{token}/sendMessage"


def send(text: str, *, chat_id: str | None = None, token: str | None = None) -> bool:
    """失敗絕不 raise；return True/False 表成功與否。"""
    token = token or get("AIDER_WATCH_TELEGRAM_TOKEN", required=True)
    chat_id = chat_id or get("AIDER_WATCH_TELEGRAM_CHAT_ID", required=True)
    url = _API.format(token=token)
    backoff = 1
    for attempt in range(3):
        try:
            r = requests.post(url, data={"chat_id": chat_id, "text": text,
                                          "parse_mode": "HTML",
                                          "disable_web_page_preview": "true"},
                              timeout=5)
            if r.status_code == 200:
                return True
            if r.status_code == 429:
                retry = r.json().get("parameters", {}).get("retry_after", backoff)
                time.sleep(retry)
                continue
        except Exception:
            pass
        time.sleep(backoff)
        backoff *= 4
    return False


def send_event(ev: _EventBase) -> bool:
    msg = format_event(ev)
    if msg is None:
        return True  # 不該推的當成功
    return send(msg)

• Step 2: 寫 integration 測試 tests/test_telegram_live.py

# 2026-04-19 @ Asia/Taipei
"""真發訊息到 DM chat_id 5619078117，訊息前綴 [TEST]。需 ~/.aider-watch.env 已設。"""
import pytest
from aider_watch.telegram import send
from aider_watch.config import get

@pytest.mark.integration
def test_real_send():
    if not get("AIDER_WATCH_TELEGRAM_TOKEN"):
        pytest.skip("TG token not configured")
    ok = send("[TEST] aider-watch integration test "
              f"(from pytest, safe to ignore)")
    assert ok is True

• Step 3: 跑 integration 測試（要求 env 已設）

pytest tests/test_telegram_live.py -v -m integration

Expected: 1 passed（Telegram 收到 [TEST] aider-watch ...）

• Step 4: Commit

git add aider_watch/telegram.py tests/test_telegram_live.py
git commit -m "feat(telegram): send() with 429 backoff + fail-open"

---

Task 6: PG Schema DDL + 建 DB 腳本

Files:

• Create: ~/aider-watch/schema.sql

• Create: ~/aider-watch/scripts/install_pg.sh

• Step 1: 寫 schema.sql

-- aider-watch PG schema | 2026-04-19 @ Asia/Taipei
-- 用在 database "aider_watch"（獨立於 awoooi 其他 DB）

CREATE TABLE IF NOT EXISTS sessions (
  id            TEXT PRIMARY KEY,
  started_at    TIMESTAMPTZ NOT NULL,
  ended_at      TIMESTAMPTZ,
  cwd           TEXT NOT NULL,
  model         TEXT NOT NULL,
  aider_args    TEXT[],
  aider_pid     INTEGER,
  duration_sec  INTEGER,
  tokens_sent   INTEGER DEFAULT 0,
  tokens_recv   INTEGER DEFAULT 0,
  cost_usd      NUMERIC(10,6) DEFAULT 0,
  files_changed INTEGER DEFAULT 0,
  error_count   INTEGER DEFAULT 0,
  exit_code     INTEGER,
  host          TEXT DEFAULT 'ogt-mac'
);
CREATE INDEX IF NOT EXISTS sessions_started_idx ON sessions(started_at DESC);
CREATE INDEX IF NOT EXISTS sessions_cwd_idx ON sessions(cwd);

CREATE TABLE IF NOT EXISTS events (
  id         BIGSERIAL PRIMARY KEY,
  session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
  ts         TIMESTAMPTZ NOT NULL,
  type       TEXT NOT NULL,
  payload    JSONB NOT NULL
);
CREATE INDEX IF NOT EXISTS events_session_ts_idx ON events(session_id, ts);
CREATE INDEX IF NOT EXISTS events_type_idx ON events(type);
CREATE INDEX IF NOT EXISTS events_ts_idx ON events(ts DESC);

CREATE TABLE IF NOT EXISTS file_touches (
  id            BIGSERIAL PRIMARY KEY,
  session_id    TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
  path          TEXT NOT NULL,
  touched_at    TIMESTAMPTZ NOT NULL,
  lines_added   INTEGER DEFAULT 0,
  lines_deleted INTEGER DEFAULT 0,
  is_new_file   BOOLEAN DEFAULT FALSE
);
CREATE INDEX IF NOT EXISTS ft_path_idx ON file_touches(path);
CREATE INDEX IF NOT EXISTS ft_session_idx ON file_touches(session_id);
CREATE INDEX IF NOT EXISTS ft_date_idx ON file_touches((touched_at::date));

• Step 2: 寫 scripts/install_pg.sh

#!/usr/bin/env bash
# aider-watch PG 初始化 | 2026-04-19 @ Asia/Taipei
# 在 192.168.0.188 建 DB + user + schema。需 peer auth（ssh 進 188 以 ogt 身份跑）。

set -euo pipefail
DB=aider_watch
USER=aider_watch
HERE="$(cd "$(dirname "$0")" && pwd)"

# 1. 生隨機密碼
PW="$(openssl rand -hex 24)"

# 2. 建 DB + user + grant（superuser 身份）
sudo -u postgres psql <<SQL
CREATE USER ${USER} WITH PASSWORD '${PW}';
CREATE DATABASE ${DB} OWNER ${USER};
GRANT ALL PRIVILEGES ON DATABASE ${DB} TO ${USER};
SQL

# 3. 套 schema
PGPASSWORD="${PW}" psql -h localhost -U "${USER}" -d "${DB}" -f "${HERE}/../schema.sql"

# 4. 輸出 DSN 給使用者（貼進 ~/.aider-watch.env）
echo "--- 貼進 Mac 的 ~/.aider-watch.env ---"
echo "AIDER_WATCH_DATABASE_URL=postgresql://${USER}:${PW}@192.168.0.188:5432/${DB}"

• Step 3: 在 188 執行（手動步驟，由統帥驗收）

scp ~/aider-watch/schema.sql ~/aider-watch/scripts/install_pg.sh ogt@192.168.0.188:/tmp/
ssh ogt@192.168.0.188 'bash /tmp/install_pg.sh'
# → 複製 AIDER_WATCH_DATABASE_URL=...

• Step 4: 驗證表存在

PGPASSWORD="..." psql -h 192.168.0.188 -U aider_watch -d aider_watch -c '\dt'

Expected: 3 tables sessions / events / file_touches

• Step 5: Commit schema & script

cd ~/aider-watch
git add schema.sql scripts/install_pg.sh
chmod +x scripts/install_pg.sh
git commit -m "feat(pg): DDL + install_pg.sh for 192.168.0.188"

---

Task 7: Storage — PG writer + JSONL buffer fallback

Files:

• Create: ~/aider-watch/aider_watch/storage.py

• Create: ~/aider-watch/tests/test_storage_pg.py

• Create: ~/aider-watch/tests/test_storage_buffer.py

• Step 1: 寫測試 test_storage_pg.py（用獨立 aider_watch_test DB）

# 2026-04-19 @ Asia/Taipei
import os, pytest, uuid
from datetime import datetime, timedelta
from aider_watch.events import SessionStart, FileEdit, SessionEnd
from aider_watch.config import TAIPEI
from aider_watch.storage import Storage

@pytest.fixture
def test_dsn():
    dsn = os.environ.get("AIDER_WATCH_TEST_DATABASE_URL")
    if not dsn:
        pytest.skip("AIDER_WATCH_TEST_DATABASE_URL not set")
    return dsn

@pytest.fixture
def store(test_dsn):
    s = Storage(dsn=test_dsn)
    s.ensure_schema()
    yield s
    s.truncate_all()

@pytest.mark.integration
def test_roundtrip_session_and_event(store):
    sid = str(uuid.uuid4())
    store.begin_session(
        session_id=sid, started_at=datetime.now(TAIPEI),
        cwd="/tmp/x", model="elephant-alpha", aider_args=["-m", "hi"],
        aider_pid=1234,
    )
    store.write_event(FileEdit(ts=datetime.now(TAIPEI), session_id=sid,
                               payload={"path": "a.py", "lines_added": 5,
                                        "lines_deleted": 1, "diff_head": [],
                                        "is_new_file": False}))
    store.end_session(session_id=sid, ended_at=datetime.now(TAIPEI),
                      duration_sec=10, tokens_sent=100, tokens_received=50,
                      cost_usd=0, files_changed=1, error_count=0, exit_code=0)

    rows = store.query("SELECT * FROM sessions WHERE id = %s", (sid,))
    assert len(rows) == 1
    assert rows[0]["model"] == "elephant-alpha"
    assert rows[0]["files_changed"] == 1

    evs = store.query("SELECT * FROM events WHERE session_id = %s", (sid,))
    assert len(evs) == 1
    assert evs[0]["type"] == "file_edit"

    ft = store.query("SELECT * FROM file_touches WHERE session_id = %s", (sid,))
    assert len(ft) == 1
    assert ft[0]["path"] == "a.py"

• Step 2: 寫測試 test_storage_buffer.py

# 2026-04-19 @ Asia/Taipei
import json
from pathlib import Path
from datetime import datetime
from aider_watch.events import FileEdit
from aider_watch.config import TAIPEI
from aider_watch.storage import Storage

def test_buffer_when_pg_offline(tmp_path, monkeypatch):
    monkeypatch.setattr("aider_watch.config.BUFFER_DIR", tmp_path)
    s = Storage(dsn="postgresql://invalid:invalid@127.0.0.1:1/nope")
    ev = FileEdit(ts=datetime.now(TAIPEI), session_id="s1",
                  payload={"path":"a","lines_added":1,"lines_deleted":0,
                           "diff_head":[],"is_new_file":False})
    s.write_event(ev)  # 不該 raise
    files = list(tmp_path.glob("pending_*.jsonl"))
    assert len(files) == 1
    line = json.loads(files[0].read_text().strip())
    assert line["type"] == "file_edit"

• Step 3: 跑驗失敗

pytest tests/test_storage_buffer.py -v

Expected: ModuleNotFoundError: No module named 'aider_watch.storage'

• Step 4: 寫 aider_watch/storage.py

# aider-watch storage | 2026-04-19 @ Asia/Taipei
"""PG writer with JSONL fallback when PG unreachable."""
from __future__ import annotations
import json, time
from pathlib import Path
from typing import Any
import psycopg2
import psycopg2.extras
from aider_watch import config
from aider_watch.events import _EventBase
from aider_watch.redactor import redact


class Storage:
    def __init__(self, dsn: str | None = None):
        self.dsn = dsn or config.get("AIDER_WATCH_DATABASE_URL")
        self._conn = None

    def _connect(self):
        if self._conn and not self._conn.closed:
            return self._conn
        self._conn = psycopg2.connect(self.dsn, connect_timeout=3)
        return self._conn

    def _disconnect(self):
        try:
            if self._conn:
                self._conn.close()
        except Exception:
            pass
        self._conn = None

    # ---- buffer ----
    def _buffer_write(self, rec: dict) -> None:
        config.ensure_dirs()
        fp = config.BUFFER_DIR / f"pending_{rec.get('session_id','unknown')}.jsonl"
        with fp.open("a") as f:
            f.write(json.dumps(rec, ensure_ascii=False, default=str) + "\n")

    # ---- public API ----
    def ensure_schema(self):
        ddl = (Path(__file__).parent.parent / "schema.sql").read_text()
        with self._connect() as c, c.cursor() as cur:
            cur.execute(ddl)
            c.commit()

    def truncate_all(self):
        with self._connect() as c, c.cursor() as cur:
            cur.execute("TRUNCATE events, file_touches, sessions RESTART IDENTITY CASCADE")
            c.commit()

    def query(self, sql: str, params=()):
        with self._connect() as c, c.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
            cur.execute(sql, params)
            return cur.fetchall() if cur.description else []

    def begin_session(self, *, session_id, started_at, cwd, model,
                       aider_args, aider_pid):
        rec = {"op":"begin_session","session_id":session_id,
               "started_at":started_at.isoformat(),"cwd":cwd,"model":model,
               "aider_args":aider_args,"aider_pid":aider_pid}
        try:
            with self._connect() as c, c.cursor() as cur:
                cur.execute(
                    "INSERT INTO sessions (id,started_at,cwd,model,aider_args,aider_pid) "
                    "VALUES (%s,%s,%s,%s,%s,%s) ON CONFLICT (id) DO NOTHING",
                    (session_id, started_at, cwd, model, aider_args, aider_pid))
                c.commit()
        except Exception:
            self._disconnect()
            self._buffer_write(rec)

    def end_session(self, *, session_id, ended_at, duration_sec, tokens_sent,
                     tokens_received, cost_usd, files_changed, error_count, exit_code):
        rec = {"op":"end_session","session_id":session_id,
               "ended_at":ended_at.isoformat(),"duration_sec":duration_sec,
               "tokens_sent":tokens_sent,"tokens_received":tokens_received,
               "cost_usd":float(cost_usd),"files_changed":files_changed,
               "error_count":error_count,"exit_code":exit_code}
        try:
            with self._connect() as c, c.cursor() as cur:
                cur.execute(
                    "UPDATE sessions SET ended_at=%s, duration_sec=%s, tokens_sent=%s, "
                    "tokens_recv=%s, cost_usd=%s, files_changed=%s, error_count=%s, "
                    "exit_code=%s WHERE id=%s",
                    (ended_at, duration_sec, tokens_sent, tokens_received,
                     cost_usd, files_changed, error_count, exit_code, session_id))
                c.commit()
        except Exception:
            self._disconnect()
            self._buffer_write(rec)

    def write_event(self, ev: _EventBase) -> None:
        payload = redact(ev.payload)
        rec = {"op":"event","session_id":ev.session_id,"type":ev.type,
               "ts":ev.ts.isoformat(),"payload":payload}
        try:
            with self._connect() as c, c.cursor() as cur:
                cur.execute(
                    "INSERT INTO events (session_id,ts,type,payload) VALUES (%s,%s,%s,%s)",
                    (ev.session_id, ev.ts, ev.type, json.dumps(payload, ensure_ascii=False)))
                if ev.type == "file_edit":
                    cur.execute(
                        "INSERT INTO file_touches (session_id,path,touched_at,lines_added,lines_deleted,is_new_file) "
                        "VALUES (%s,%s,%s,%s,%s,%s)",
                        (ev.session_id, payload["path"], ev.ts,
                         payload.get("lines_added",0), payload.get("lines_deleted",0),
                         payload.get("is_new_file", False)))
                c.commit()
        except Exception:
            self._disconnect()
            self._buffer_write(rec)

    def flush_buffer(self) -> int:
        """把 buffer 下所有 pending_*.jsonl 重新套用；成功一行刪一行（原檔重寫）。回傳成功筆數。"""
        config.ensure_dirs()
        applied = 0
        for fp in list(config.BUFFER_DIR.glob("pending_*.jsonl")):
            lines = fp.read_text().splitlines()
            remaining = []
            for line in lines:
                if not line.strip():
                    continue
                rec = json.loads(line)
                try:
                    self._apply(rec)
                    applied += 1
                except Exception:
                    remaining.append(line)  # 保留失敗的
            if remaining:
                fp.write_text("\n".join(remaining) + "\n")
            else:
                fp.unlink()
        return applied

    def _apply(self, rec: dict):
        op = rec.get("op")
        if op == "event":
            with self._connect() as c, c.cursor() as cur:
                cur.execute(
                    "INSERT INTO events (session_id,ts,type,payload) VALUES (%s,%s,%s,%s)",
                    (rec["session_id"], rec["ts"], rec["type"],
                     json.dumps(rec["payload"], ensure_ascii=False)))
                if rec["type"] == "file_edit":
                    p = rec["payload"]
                    cur.execute(
                        "INSERT INTO file_touches (session_id,path,touched_at,lines_added,lines_deleted,is_new_file) "
                        "VALUES (%s,%s,%s,%s,%s,%s)",
                        (rec["session_id"], p["path"], rec["ts"],
                         p.get("lines_added",0), p.get("lines_deleted",0),
                         p.get("is_new_file", False)))
                c.commit()
        elif op == "begin_session":
            with self._connect() as c, c.cursor() as cur:
                cur.execute(
                    "INSERT INTO sessions (id,started_at,cwd,model,aider_args,aider_pid) "
                    "VALUES (%s,%s,%s,%s,%s,%s) ON CONFLICT (id) DO NOTHING",
                    (rec["session_id"], rec["started_at"], rec["cwd"],
                     rec["model"], rec["aider_args"], rec["aider_pid"]))
                c.commit()
        elif op == "end_session":
            with self._connect() as c, c.cursor() as cur:
                cur.execute(
                    "UPDATE sessions SET ended_at=%s, duration_sec=%s, tokens_sent=%s, "
                    "tokens_recv=%s, cost_usd=%s, files_changed=%s, error_count=%s, "
                    "exit_code=%s WHERE id=%s",
                    (rec["ended_at"], rec["duration_sec"], rec["tokens_sent"],
                     rec["tokens_received"], rec["cost_usd"], rec["files_changed"],
                     rec["error_count"], rec["exit_code"], rec["session_id"]))
                c.commit()

• Step 5: 跑測試驗通過（需先設 AIDER_WATCH_TEST_DATABASE_URL 指向 aider_watch_test）

# 在 188 預先建 aider_watch_test DB：
ssh ogt@192.168.0.188 "sudo -u postgres createdb -O aider_watch aider_watch_test"
export AIDER_WATCH_TEST_DATABASE_URL="postgresql://aider_watch:<pw>@192.168.0.188:5432/aider_watch_test"
pytest tests/test_storage_pg.py tests/test_storage_buffer.py -v -m integration

Expected: 2 passed

• Step 6: Commit

git add aider_watch/storage.py tests/test_storage_pg.py tests/test_storage_buffer.py
git commit -m "feat(storage): PG writer + JSONL buffer fallback + flush"

---

Task 8: aider stdout / chat_history / git diff 解析

Files:

• Create: ~/aider-watch/aider_watch/parsers.py

• Create: ~/aider-watch/tests/test_parsers.py

• Create: ~/aider-watch/tests/fixtures/aider_stdout_sample.txt

• Create: ~/aider-watch/tests/fixtures/aider_chat_history_sample.md

• Step 1: 先跑一次真 aider 抓 stdout 樣本

cd /tmp && rm -rf pa && mkdir pa && cd pa && git init -q
echo "def add(a,b): return 0" > calc.py && git add -A && git commit -m init -q
aider --model openrouter/openrouter/elephant-alpha --yes --no-stream --no-pretty \
  --message "fix calc.py add function — should return a+b" --exit \
  > ~/aider-watch/tests/fixtures/aider_stdout_sample.txt 2>&1 || true
cp .aider.chat.history.md ~/aider-watch/tests/fixtures/aider_chat_history_sample.md

• Step 2: 寫測試 tests/test_parsers.py（讀 fixture 驗解析）

# 2026-04-19 @ Asia/Taipei
from pathlib import Path
from aider_watch.parsers import parse_stdout_banner, parse_chat_history, parse_git_diff_stat

FIX = Path(__file__).parent / "fixtures"

def test_parse_banner():
    out = (FIX / "aider_stdout_sample.txt").read_text()
    info = parse_stdout_banner(out)
    assert info["model"] is not None and "elephant" in info["model"].lower()

def test_parse_chat_history_token_totals():
    md = (FIX / "aider_chat_history_sample.md").read_text()
    stats = parse_chat_history(md)
    assert stats["tokens_sent"] >= 0
    assert stats["tokens_received"] >= 0
    assert isinstance(stats["file_edits"], list)

def test_parse_git_diff_stat_empty():
    # 空 diff = 沒改檔
    assert parse_git_diff_stat("") == []

• Step 3: 跑驗失敗

pytest tests/test_parsers.py -v

Expected: ModuleNotFoundError: No module named 'aider_watch.parsers'

• Step 4: 寫 aider_watch/parsers.py

# aider-watch parsers | 2026-04-19 @ Asia/Taipei
"""從 aider stdout / chat_history.md / git diff 抽事件。"""
from __future__ import annotations
import re, subprocess
from pathlib import Path

_BANNER_MODEL = re.compile(r"^\s*Model:\s*(\S+)", re.MULTILINE)
_BANNER_VER = re.compile(r"^\s*Aider v([\d\.]+)", re.MULTILINE)
_TOKENS_LINE = re.compile(r"Tokens:\s+([\d\.]+)([km]?)\s+sent,\s+([\d\.]+)([km]?)\s+received", re.IGNORECASE)


def _parse_num(n: str, suffix: str) -> int:
    v = float(n)
    if suffix.lower() == "k":
        v *= 1000
    elif suffix.lower() == "m":
        v *= 1_000_000
    return int(v)


def parse_stdout_banner(stdout: str) -> dict:
    m_model = _BANNER_MODEL.search(stdout)
    m_ver = _BANNER_VER.search(stdout)
    return {
        "model": m_model.group(1) if m_model else None,
        "cli_version": m_ver.group(1) if m_ver else None,
    }


def parse_chat_history(md: str) -> dict:
    """從 .aider.chat.history.md 抽 token 累計 + 檔案提及。"""
    total_sent = 0
    total_recv = 0
    for m in _TOKENS_LINE.finditer(md):
        total_sent += _parse_num(m.group(1), m.group(2))
        total_recv += _parse_num(m.group(3), m.group(4))
    # 檔案提及：>>>>>>> REPLACE 之前的 block 頭有檔名，這裡先簡版，用 file_edit 回 git diff 為主
    file_edits: list[str] = []  # 留給後續增強，canonical 以 git diff 為準
    return {"tokens_sent": total_sent, "tokens_received": total_recv,
            "file_edits": file_edits}


def parse_git_diff_stat(diff_stat: str) -> list[dict]:
    """解析 `git diff --numstat <pre>..<post>` 輸出：\"12\\t3\\tpath/to/file\"。"""
    out = []
    for line in diff_stat.strip().splitlines():
        parts = line.split("\t")
        if len(parts) != 3:
            continue
        add, dele, path = parts
        try:
            a = int(add) if add != "-" else 0
            d = int(dele) if dele != "-" else 0
        except ValueError:
            continue
        out.append({"path": path, "lines_added": a, "lines_deleted": d,
                    "is_new_file": d == 0 and not Path(path).exists() is False})
    return out


def git_numstat(cwd: Path, pre_sha: str, post_sha: str = "HEAD") -> list[dict]:
    """從 repo 取 diff numstat。若 cwd 不是 repo 或 pre_sha 無效回空 list。"""
    try:
        r = subprocess.run(
            ["git", "-C", str(cwd), "diff", "--numstat", pre_sha, post_sha],
            capture_output=True, text=True, timeout=10)
        if r.returncode != 0:
            return []
        return parse_git_diff_stat(r.stdout)
    except Exception:
        return []


def git_diff_head(cwd: Path, path: str, pre_sha: str, n: int = 3) -> list[str]:
    """取某檔 diff 前 n 行（給 Telegram 預覽用）。"""
    try:
        r = subprocess.run(
            ["git", "-C", str(cwd), "diff", "-U0", pre_sha, "--", path],
            capture_output=True, text=True, timeout=5)
        lines = r.stdout.splitlines()
        picks = []
        for l in lines:
            if l.startswith("@@") or l.startswith("+") or l.startswith("-"):
                if l.startswith("+++") or l.startswith("---"):
                    continue
                picks.append(l)
                if len(picks) >= n:
                    break
        return picks
    except Exception:
        return []

• Step 5: 跑測試驗通過

pytest tests/test_parsers.py -v

Expected: 3 passed

• Step 6: Commit

git add aider_watch/parsers.py tests/test_parsers.py tests/fixtures/
git commit -m "feat(parsers): stdout banner + chat_history tokens + git numstat"

---

Task 9: Wrapper — subprocess 生命週期 + Dispatcher

Files:

• Create: ~/aider-watch/aider_watch/wrapper.py

• Create: ~/aider-watch/bin/aiderw

• Step 1: 寫 aider_watch/wrapper.py

# aider-watch wrapper | 2026-04-19 @ Asia/Taipei
"""aider subprocess wrapper：stdout tee、即時 event、退出時 canonical 重建。"""
from __future__ import annotations
import atexit, json, os, socket, subprocess, sys, threading, time
from datetime import datetime
from pathlib import Path
from ulid import ULID
from aider_watch import config
from aider_watch.config import TAIPEI, SESSIONS_DIR, LIVE_LOG
from aider_watch.events import (
    SessionStart, FileEdit, ErrorEvent, SilentTimeout, SessionEnd, _EventBase)
from aider_watch.storage import Storage
from aider_watch.telegram import send_event
from aider_watch.parsers import parse_stdout_banner, parse_chat_history, git_numstat, git_diff_head

IDLE_THRESHOLD_SEC = 120


class Dispatcher:
    def __init__(self, store: Storage, session_id: str, session_jsonl: Path):
        self.store = store
        self.session_id = session_id
        self.jsonl = session_jsonl
        self.jsonl.parent.mkdir(parents=True, exist_ok=True)

    def emit(self, ev: _EventBase) -> None:
        # 1. JSONL source-of-truth
        with self.jsonl.open("a") as f:
            f.write(ev.model_dump_json() + "\n")
        # 2. live.log
        try:
            with LIVE_LOG.open("a") as f:
                f.write(f"[{ev.ts.isoformat()}] {ev.type} {json.dumps(ev.payload, ensure_ascii=False, default=str)[:200]}\n")
        except Exception:
            pass
        # 3. PG (with fallback)
        try:
            self.store.write_event(ev)
        except Exception:
            pass
        # 4. Telegram
        try:
            send_event(ev)
        except Exception:
            pass


def _now() -> datetime:
    return datetime.now(TAIPEI)


def _pre_git_sha(cwd: Path) -> str | None:
    try:
        r = subprocess.run(["git", "-C", str(cwd), "rev-parse", "HEAD"],
                           capture_output=True, text=True, timeout=3)
        return r.stdout.strip() if r.returncode == 0 else None
    except Exception:
        return None


def main(argv: list[str] | None = None) -> int:
    argv = list(argv if argv is not None else sys.argv[1:])
    config.ensure_dirs()
    cwd = Path.cwd()

    sid = str(ULID())
    started = _now()
    date_dir = SESSIONS_DIR / started.strftime("%Y/%m/%d")
    jsonl = date_dir / f"{sid}_{cwd.name}_{os.getpid()}.jsonl"
    stdout_buf: list[str] = []
    last_output_ts = [time.time()]
    error_count = [0]
    wrapper_crashed = [False]
    pre_sha = _pre_git_sha(cwd)

    store = Storage()
    dispatch = Dispatcher(store, sid, jsonl)

    # 找 aider 實體（不包自己）
    aider_bin = os.environ.get("AIDER_BIN") or str(Path.home() / ".local/bin/aider")

    # 啟 subprocess
    proc = subprocess.Popen(
        [aider_bin] + argv, cwd=str(cwd),
        stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
        text=True, bufsize=1,  # line-buffered
    )

    # 即時發 session_start
    store.begin_session(
        session_id=sid, started_at=started, cwd=str(cwd),
        model="openrouter/openrouter/elephant-alpha",  # default; parse_stdout_banner 後更新
        aider_args=argv, aider_pid=proc.pid,
    )
    dispatch.emit(SessionStart(
        ts=started, session_id=sid,
        payload={"cwd": str(cwd), "model": "openrouter/openrouter/elephant-alpha",
                 "aider_args": argv, "aider_pid": proc.pid, "cli_version": "unknown"}))

    # stdout 讀取 thread + tee
    def _reader():
        for line in proc.stdout:  # type: ignore[arg-type]
            sys.stdout.write(line)
            sys.stdout.flush()
            stdout_buf.append(line)
            last_output_ts[0] = time.time()
            if "error" in line.lower() or "exception" in line.lower():
                error_count[0] += 1
    t = threading.Thread(target=_reader, daemon=True)
    t.start()

    # silent timeout monitor
    silent_fired = [False]
    def _silent_watch():
        while proc.poll() is None:
            time.sleep(10)
            idle = time.time() - last_output_ts[0]
            if idle >= IDLE_THRESHOLD_SEC and not silent_fired[0]:
                silent_fired[0] = True
                tail = "".join(stdout_buf[-1:])[-80:] if stdout_buf else ""
                dispatch.emit(SilentTimeout(
                    ts=_now(), session_id=sid,
                    payload={"idle_sec": int(idle), "last_output_tail": tail}))
    w = threading.Thread(target=_silent_watch, daemon=True)
    w.start()

    # atexit 保險：wrapper 崩潰時仍寫半份 session_end
    def _emergency():
        if wrapper_crashed[0]:
            return  # 已正常結束
        try:
            store.end_session(
                session_id=sid, ended_at=_now(), duration_sec=int(time.time()),
                tokens_sent=0, tokens_received=0, cost_usd=0,
                files_changed=0, error_count=error_count[0], exit_code=-999)
        except Exception:
            pass
    atexit.register(_emergency)

    # 等結束
    exit_code = proc.wait()
    wrapper_crashed[0] = True  # 正常路徑，關掉 atexit 保險
    duration = int((_now() - started).total_seconds())

    # 銜接 stdout banner 資訊
    stdout_text = "".join(stdout_buf)
    banner = parse_stdout_banner(stdout_text)

    # canonical file edits from git diff
    if pre_sha:
        edits = git_numstat(cwd, pre_sha, "HEAD")
        for e in edits:
            head = git_diff_head(cwd, e["path"], pre_sha, 3)
            dispatch.emit(FileEdit(
                ts=_now(), session_id=sid,
                payload={"path": e["path"], "lines_added": e["lines_added"],
                         "lines_deleted": e["lines_deleted"], "diff_head": head,
                         "is_new_file": e["is_new_file"]}))

    # tokens 從 chat_history.md
    tokens_sent = tokens_recv = 0
    chat_md = cwd / ".aider.chat.history.md"
    if chat_md.exists():
        s = parse_chat_history(chat_md.read_text(errors="ignore"))
        tokens_sent = s["tokens_sent"]
        tokens_recv = s["tokens_received"]

    # error event 若 exit code 非 0
    if exit_code != 0:
        dispatch.emit(ErrorEvent(
            ts=_now(), session_id=sid,
            payload={"kind": "non_zero_exit",
                     "message": f"aider exited with {exit_code}",
                     "context_50chars": stdout_text[-50:]}))

    files_changed = len(git_numstat(cwd, pre_sha, "HEAD")) if pre_sha else 0

    # 最終 session_end
    dispatch.emit(SessionEnd(
        ts=_now(), session_id=sid,
        payload={"duration_sec": duration, "tokens_sent": tokens_sent,
                 "tokens_received": tokens_recv, "cost_usd": 0,
                 "files_changed": files_changed, "error_count": error_count[0],
                 "exit_code": exit_code}))
    store.end_session(
        session_id=sid, ended_at=_now(), duration_sec=duration,
        tokens_sent=tokens_sent, tokens_received=tokens_recv, cost_usd=0,
        files_changed=files_changed, error_count=error_count[0], exit_code=exit_code)

    return exit_code


if __name__ == "__main__":
    sys.exit(main())

• Step 2: 寫 bin/aiderw（shim）

#!/usr/bin/env python3
# aider-watch shim | 2026-04-19 @ Asia/Taipei
import sys
from aider_watch.wrapper import main
sys.exit(main())

chmod +x ~/aider-watch/bin/aiderw

• Step 3: Commit

git add aider_watch/wrapper.py bin/aiderw
git commit -m "feat(wrapper): subprocess + stdout tee + silent timeout + canonical rebuild"

---

Task 10: CLI — aider-watch {doctor,flush,replay,report}

Files:

• Create: ~/aider-watch/aider_watch/cli.py

• Step 1: 寫 aider_watch/cli.py

# aider-watch CLI | 2026-04-19 @ Asia/Taipei
from __future__ import annotations
import argparse, json, os, subprocess, sys
from pathlib import Path
from datetime import datetime
from aider_watch import config
from aider_watch.config import TAIPEI, LOGS_DIR, STATE_FILE
from aider_watch.storage import Storage


def cmd_doctor(_args) -> int:
    ok = True
    print("== aider-watch doctor ==")
    # 1. env
    for k in ("AIDER_WATCH_DATABASE_URL", "AIDER_WATCH_TELEGRAM_TOKEN",
              "AIDER_WATCH_TELEGRAM_CHAT_ID"):
        v = config.get(k)
        status = "✅" if v else "❌"
        print(f"  env {k:40s} {status}")
        if not v: ok = False
    # 2. PG
    try:
        s = Storage()
        r = s.query("SELECT count(*) AS c FROM sessions")
        print(f"  PG reachable ✅  sessions={r[0]['c']}")
    except Exception as e:
        print(f"  PG reachable ❌  {e}")
        ok = False
    # 3. Telegram（getMe，不發訊息）
    try:
        import requests
        tok = config.get("AIDER_WATCH_TELEGRAM_TOKEN", required=True)
        r = requests.get(f"https://api.telegram.org/bot{tok}/getMe", timeout=5)
        if r.status_code == 200 and r.json().get("ok"):
            print(f"  Telegram bot ✅ @{r.json()['result']['username']}")
        else:
            print(f"  Telegram bot ❌ {r.status_code}"); ok = False
    except Exception as e:
        print(f"  Telegram bot ❌ {e}"); ok = False
    # 4. launchd
    res = subprocess.run(["launchctl", "list"], capture_output=True, text=True)
    for label in ("com.awoooi.aider-watch.flush",
                  "com.awoooi.aider-watch.daily",
                  "com.awoooi.aider-watch.weekly"):
        mark = "✅" if label in res.stdout else "❌"
        print(f"  launchd {label:40s} {mark}")
        if mark == "❌": ok = False
    # 5. 磁碟
    import shutil
    du = shutil.disk_usage(str(config.WATCH_ROOT))
    print(f"  disk free: {du.free/1e9:.1f}G / {du.total/1e9:.1f}G")
    return 0 if ok else 1


def cmd_flush(_args) -> int:
    s = Storage()
    n = s.flush_buffer()
    STATE_FILE.write_text(json.dumps({"last_flush_ts": datetime.now(TAIPEI).isoformat(),
                                       "applied": n}))
    print(f"flushed {n} records")
    return 0


def cmd_replay(args) -> int:
    """從本機 JSONL 重建 events 到 PG（幂等依 ON CONFLICT DO NOTHING）。"""
    sid = args.session_id
    hits = list(config.SESSIONS_DIR.rglob(f"{sid}_*.jsonl"))
    if not hits:
        print(f"session {sid} not found in sessions/"); return 1
    from aider_watch.events import from_json
    s = Storage()
    for fp in hits:
        for line in fp.read_text().splitlines():
            if not line.strip(): continue
            ev = from_json(line)
            s.write_event(ev)
    print(f"replayed {sid}")
    return 0


def cmd_report(args) -> int:
    from aider_watch.reporter import daily, weekly
    if args.kind == "daily": return daily()
    if args.kind == "weekly": return weekly()
    print("unknown report kind"); return 1


def main() -> int:
    p = argparse.ArgumentParser(prog="aider-watch")
    sub = p.add_subparsers(dest="cmd", required=True)
    sub.add_parser("doctor").set_defaults(func=cmd_doctor)
    sub.add_parser("flush").set_defaults(func=cmd_flush)
    rp = sub.add_parser("replay"); rp.add_argument("session_id"); rp.set_defaults(func=cmd_replay)
    rr = sub.add_parser("report"); rr.add_argument("kind", choices=["daily","weekly"]); rr.set_defaults(func=cmd_report)
    a = p.parse_args()
    return a.func(a)


if __name__ == "__main__":
    sys.exit(main())

• Step 2: Commit

git add aider_watch/cli.py
git commit -m "feat(cli): doctor/flush/replay/report subcommands"

---

Task 11: Reporter — daily/weekly aggregation

Files:

• Create: ~/aider-watch/aider_watch/reporter.py

• Step 1: 寫 aider_watch/reporter.py

# aider-watch reporter | 2026-04-19 @ Asia/Taipei
from __future__ import annotations
from datetime import datetime, timedelta
from aider_watch.config import TAIPEI
from aider_watch.storage import Storage
from aider_watch.telegram import send


def _fmt_dur(sec: int) -> str:
    h, rem = divmod(sec, 3600); m, s = divmod(rem, 60)
    if h: return f"{h}h{m}m"
    if m: return f"{m}m{s}s"
    return f"{s}s"


def _fmt_tokens(n: int) -> str:
    return f"{n/1000:.1f}k" if n >= 1000 else str(n)


def _render(title: str, start, stats: dict, top_files: list[dict]) -> str:
    lines = [f"📊 <b>{title}</b>  <i>({start})</i>\n"]
    if stats["sessions"] == 0:
        lines.append("（無 aider 活動）")
        return "\n".join(lines)
    lines.append(f"• Sessions: {stats['sessions']}")
    lines.append(f"• 總時長: {_fmt_dur(stats['total_sec'] or 0)}")
    lines.append(f"• Tokens 送/收: {_fmt_tokens(stats['toks_sent'] or 0)} / {_fmt_tokens(stats['toks_recv'] or 0)}")
    lines.append(f"• 檔案變更: {stats['files'] or 0}")
    lines.append(f"• 錯誤: {stats['errors'] or 0}")
    if top_files:
        lines.append("\n<b>Top 檔案</b>:")
        for f in top_files[:5]:
            lines.append(f"  • <code>{f['path']}</code>  {f['touches']}× (+{f['adds']}/-{f['dels']})")
    return "\n".join(lines)


def _report(start_dt: datetime, end_dt: datetime, title: str) -> int:
    s = Storage()
    row = s.query(
        "SELECT count(*) AS sessions, sum(duration_sec) AS total_sec, "
        "sum(tokens_sent) AS toks_sent, sum(tokens_recv) AS toks_recv, "
        "sum(files_changed) AS files, sum(error_count) AS errors "
        "FROM sessions WHERE started_at >= %s AND started_at < %s",
        (start_dt, end_dt))[0]
    top = s.query(
        "SELECT path, count(*) AS touches, sum(lines_added) AS adds, "
        "sum(lines_deleted) AS dels "
        "FROM file_touches WHERE touched_at >= %s AND touched_at < %s "
        "GROUP BY path ORDER BY touches DESC LIMIT 5",
        (start_dt, end_dt))
    msg = _render(title, start_dt.strftime("%Y-%m-%d"), row, top)
    ok = send(msg)
    return 0 if ok else 1


def daily() -> int:
    now = datetime.now(TAIPEI)
    start = now.replace(hour=0, minute=0, second=0, microsecond=0)
    end = start + timedelta(days=1)
    return _report(start, end, "aider 日報")


def weekly() -> int:
    now = datetime.now(TAIPEI)
    start = now.replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=now.weekday())
    end = start + timedelta(days=7)
    return _report(start, end, "aider 週報")

• Step 2: Commit

git add aider_watch/reporter.py
git commit -m "feat(reporter): daily/weekly aggregation → TG"

---

Task 12: Launchd plist 3 支

Files:

• Create: ~/aider-watch/launchd/com.awoooi.aider-watch.flush.plist

• Create: ~/aider-watch/launchd/com.awoooi.aider-watch.daily.plist

• Create: ~/aider-watch/launchd/com.awoooi.aider-watch.weekly.plist

• Create: ~/aider-watch/scripts/install_launchd.sh

• Step 1: 寫 flush.plist（每 5 分鐘）

<?xml version="1.0" encoding="UTF-8"?>
<!-- aider-watch flush | 2026-04-19 @ Asia/Taipei -->
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0"><dict>
  <key>Label</key><string>com.awoooi.aider-watch.flush</string>
  <key>ProgramArguments</key>
  <array>
    <string>/Users/ogt/.local/bin/aider-watch</string>
    <string>flush</string>
  </array>
  <key>StartInterval</key><integer>300</integer>
  <key>EnvironmentVariables</key>
  <dict><key>PATH</key><string>/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin</string></dict>
  <key>StandardOutPath</key><string>/Users/ogt/aider-watch/logs/flush.log</string>
  <key>StandardErrorPath</key><string>/Users/ogt/aider-watch/logs/flush.log</string>
  <key>KeepAlive</key><false/>
  <key>RunAtLoad</key><false/>
</dict></plist>

• Step 2: 寫 daily.plist（每日 23:50）

<?xml version="1.0" encoding="UTF-8"?>
<!-- aider-watch daily | 2026-04-19 @ Asia/Taipei -->
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0"><dict>
  <key>Label</key><string>com.awoooi.aider-watch.daily</string>
  <key>ProgramArguments</key>
  <array>
    <string>/Users/ogt/.local/bin/aider-watch</string>
    <string>report</string><string>daily</string>
  </array>
  <key>StartCalendarInterval</key>
  <dict><key>Hour</key><integer>23</integer><key>Minute</key><integer>50</integer></dict>
  <key>EnvironmentVariables</key>
  <dict><key>PATH</key><string>/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin</string></dict>
  <key>StandardOutPath</key><string>/Users/ogt/aider-watch/logs/daily.log</string>
  <key>StandardErrorPath</key><string>/Users/ogt/aider-watch/logs/daily.log</string>
</dict></plist>

• Step 3: 寫 weekly.plist（週日 22:00）

<?xml version="1.0" encoding="UTF-8"?>
<!-- aider-watch weekly | 2026-04-19 @ Asia/Taipei -->
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0"><dict>
  <key>Label</key><string>com.awoooi.aider-watch.weekly</string>
  <key>ProgramArguments</key>
  <array>
    <string>/Users/ogt/.local/bin/aider-watch</string>
    <string>report</string><string>weekly</string>
  </array>
  <key>StartCalendarInterval</key>
  <dict><key>Weekday</key><integer>0</integer>
        <key>Hour</key><integer>22</integer><key>Minute</key><integer>0</integer></dict>
  <key>EnvironmentVariables</key>
  <dict><key>PATH</key><string>/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin</string></dict>
  <key>StandardOutPath</key><string>/Users/ogt/aider-watch/logs/weekly.log</string>
  <key>StandardErrorPath</key><string>/Users/ogt/aider-watch/logs/weekly.log</string>
</dict></plist>

• Step 4: 寫 scripts/install_launchd.sh

#!/usr/bin/env bash
# aider-watch launchd install | 2026-04-19 @ Asia/Taipei
set -euo pipefail
HERE="$(cd "$(dirname "$0")" && pwd)"
DEST="$HOME/Library/LaunchAgents"
mkdir -p "$DEST" "$HOME/aider-watch/logs"

for p in flush daily weekly; do
  cp "${HERE}/../launchd/com.awoooi.aider-watch.${p}.plist" "${DEST}/"
  launchctl unload "${DEST}/com.awoooi.aider-watch.${p}.plist" 2>/dev/null || true
  launchctl load -w "${DEST}/com.awoooi.aider-watch.${p}.plist"
  echo "loaded: com.awoooi.aider-watch.${p}"
done
launchctl list | grep aider-watch

• Step 5: Commit

chmod +x scripts/install_launchd.sh
git add launchd/ scripts/install_launchd.sh
git commit -m "feat(launchd): flush 5min + daily 23:50 + weekly Sun 22:00"

---

Task 13: 總安裝腳本

Files:

• Create: ~/aider-watch/scripts/install.sh

• Step 1: 寫 scripts/install.sh

#!/usr/bin/env bash
# aider-watch one-shot installer | 2026-04-19 @ Asia/Taipei
set -euo pipefail
REPO="$HOME/aider-watch"
cd "$REPO"

# 1. 套件
pipx install --python /Users/ogt/.pyenv/versions/3.11.7/bin/python3 -e . || pipx upgrade aider-watch
which aiderw aider-watch

# 2. symlink 入 PATH
ln -sf "$HOME/.local/bin/aiderw" /opt/homebrew/bin/aiderw
echo "installed: aiderw → /opt/homebrew/bin/aiderw"

# 3. env 檢查（使用者需已建）
if [[ ! -f "$HOME/.aider-watch.env" ]]; then
  echo "❌ 請先建 ~/.aider-watch.env（至少 AIDER_WATCH_DATABASE_URL/TELEGRAM_TOKEN/CHAT_ID）"
  exit 1
fi
chmod 600 "$HOME/.aider-watch.env"

# 4. 工作目錄
mkdir -p "$HOME/aider-watch/"{sessions,buffer,logs}

# 5. launchd
bash "$REPO/scripts/install_launchd.sh"

# 6. doctor
aider-watch doctor

• Step 2: Commit

chmod +x scripts/install.sh
git add scripts/install.sh
git commit -m "feat(install): one-shot installer"

---

Task 14: 寫 ~/.aider-watch.env + 跑 install + E2E

Files:

• Create: ~/.aider-watch.env（手動，含 Task 6 Step 3 輸出的 DSN）

• Step 1: 寫 ~/.aider-watch.env（chmod 600）

cat > ~/.aider-watch.env <<'EOF'
# aider-watch secrets | 2026-04-19 @ Asia/Taipei
AIDER_WATCH_DATABASE_URL=postgresql://aider_watch:<Task6產生的PW>@192.168.0.188:5432/aider_watch
AIDER_WATCH_TELEGRAM_TOKEN=8474499448:AAFqu_i4-PN4zGFOK5ea8o0Ud56qqEtCMeI
AIDER_WATCH_TELEGRAM_CHAT_ID=5619078117
AIDER_WATCH_HOSTNAME=ogt-mac
EOF
chmod 600 ~/.aider-watch.env

• Step 2: 跑安裝

cd ~/aider-watch && bash scripts/install.sh

Expected: aider-watch doctor 全 ✅

• Step 3: E2E — happy path

mkdir -p /tmp/aw-e2e && cd /tmp/aw-e2e
git init -q && echo "def f(): return 0" > calc.py && git add -A && git commit -m init -q
aiderw --yes --no-stream --no-pretty \
  --message "fix calc.py f should return 42" --exit

Expected:

• Telegram 收到 🚀 + ✏️ (或 🆕) + 🏁（若改檔；若沒改也至少有 🚀 + 🏁）

• psql ... 查 sessions 有 1 筆 exit_code=0

• ~/aider-watch/sessions/2026/04/19/<sid>_aw-e2e_<pid>.jsonl 存在

• Step 4: E2E — PG 斷線降級

# 模擬：用錯的 DSN 跑一次
export AIDER_WATCH_DATABASE_URL="postgresql://x:x@127.0.0.1:1/nope"
cd /tmp/aw-e2e && aiderw --yes --no-stream --no-pretty --message "add comment to calc.py" --exit
ls ~/aider-watch/buffer/  # → 應有 pending_*.jsonl
unset AIDER_WATCH_DATABASE_URL
aider-watch flush
ls ~/aider-watch/buffer/  # → 應為空

Expected: buffer 產生 → flush 後 PG 補齊、buffer 清空

• Step 5: E2E — 日報

aider-watch report daily

Expected: Telegram 收到日報，數字含今天的 2 場 session

• Step 6: 最終 commit

cd ~/aider-watch
git add -A
git commit --allow-empty -m "chore: E2E verified 2026-04-19"

---

Self-Review 結論

Spec 覆蓋檢查（設計稿 §1-§7 逐段）：

設計稿章節	對應 Task
§1 架構總覽	Task 9 (wrapper), 11 (reporter), 12 (launchd)
§2 元件分工 5 個單元	Task 1 (events), 2+4+5 (redactor/tg), 7 (storage), 8 (parsers), 9 (wrapper), 10 (cli), 11 (reporter)
§3 Event Schema 7 種	Task 1 全部 7 類
§4 PG schema 3 張表	Task 6
§5 檔案佈局	Task 0 (scaffold), 3 (config), 12 (launchd), 13 (install), 14 (env)
§6 降級：PG/TG/crash/secret	Task 5 (TG backoff), 7 (PG fallback), 9 (atexit), 2 (redactor)
§7 測試 unit/integration/E2E	Task 1,2,4,8 (unit); 5,7 (integration); 14 (E2E 三種)

沒有漏。

Placeholder 掃描：無「TBD / TODO / 略」。<Task6產生的PW> 是 Task 6 Step 3 實際生成的真實值，執行時填入，非設計漏洞。

Type 一致性：event 欄位名稱 tokens_sent/tokens_received 在 events.py → telegram.py → storage.py → reporter.py 全統一（storage 寫入 PG 時映射到欄位 tokens_recv 是 PG DDL 名，程式內仍用 tokens_received 傳參）。

---

Execution Handoff

Plan complete and saved to docs/superpowers/plans/2026-04-19-aider-watch.md。

兩種執行方式：

1. Subagent-Driven（推薦） — 每個 Task 派一個獨立 subagent 跑，Claude 逐任務 review、測試、commit 再進下一個。優點：隔離、失敗不汙染 context、每 task 有 review gate
2. Inline Execution — 本 session 逐步跑，批次到 checkpoint 停一下讓你 review

選哪個？