1
#!/usr/bin/env python3
2
from __future__ import annotations
3

4
import argparse
5
import os
6
import re
7
import sqlite3
8
import sys
9
import time
10
from datetime import datetime
11
from pathlib import Path
12

13
TRIGGER_NAME = "codex_drop_low_value_logs"
14
BLOCKED_LEVELS = ("TRACE", "DEBUG", "INFO")
15
KNOWN_TRIGGERS = (
16
    TRIGGER_NAME,
17
    "codex_keep_warn_error_logs",
18
    "block_noisy_logs",
19
    "codex_drop_trace_logs",
20
    "codex_drop_all_logs",
21
)
22
TRIGGER_SQL = f"""
23
CREATE TRIGGER {TRIGGER_NAME}
24
BEFORE INSERT ON logs
25
WHEN upper(coalesce(NEW.level, '')) IN ('TRACE', 'DEBUG', 'INFO')
26
BEGIN
27
    SELECT RAISE(IGNORE);
28
END
29
""".strip()
30

31
def fail(message: str) -> None:
32
    raise SystemExit(f"error: {message}")
33

34
def default_db() -> Path:
35
    override = os.environ.get("CODEX_LOG_DB")
36
    return Path(override).expanduser() if override else Path.home() / ".codex" / "logs_2.sqlite"
37

38
def connect(db: Path, readonly: bool) -> sqlite3.Connection:
39
    if not db.exists():
40
        fail(f"database not found: {db}")
41
    if readonly:
42
        con = sqlite3.connect(db.resolve().as_uri() + "?mode=ro", uri=True, timeout=30)
43
    else:
44
        con = sqlite3.connect(db, timeout=30)
45
    con.row_factory = sqlite3.Row
46
    con.execute("PRAGMA busy_timeout=30000")
47
    return con
48

49
def assert_schema(con: sqlite3.Connection) -> None:
50
    table = con.execute(
51
        "SELECT 1 FROM sqlite_master WHERE type='table' AND name='logs'"
52
    ).fetchone()
53
    if not table:
54
        fail("logs table not found")
55
    columns = {row["name"] for row in con.execute("PRAGMA table_info(logs)")}
56
    missing = {"id", "ts", "level"} - columns
57
    if missing:
58
        fail(f"logs table missing columns: {', '.join(sorted(missing))}")
59

60
def normalized(sql: str) -> str:
61
    return re.sub(r"\s+", "", sql).rstrip(";").lower()
62

63
def trigger_ok(con: sqlite3.Connection) -> bool:
64
    row = con.execute(
65
        "SELECT sql FROM sqlite_master WHERE type='trigger' AND name=?",
66
        (TRIGGER_NAME,),
67
    ).fetchone()
68
    return bool(row and normalized(row["sql"]) == normalized(TRIGGER_SQL))
69

70
def backup(db: Path) -> Path:
71
    stamp = datetime.now().strftime("%Y%m%d-%H%M%S-%f")
72
    target = db.parent / f"{db.name}.backup-{stamp}.sqlite"
73
    with sqlite3.connect(db, timeout=30) as src, sqlite3.connect(target) as dst:
74
        src.backup(dst)
75
    print(f"backup: {target} ({target.stat().st_size} bytes)")
76
    return target
77

78
def status(db: Path, seconds: int) -> None:
79
    with connect(db, readonly=True) as con:
80
        assert_schema(con)
81
        columns = {row["name"] for row in con.execute("PRAGMA table_info(logs)")}
82
        byte_expr = (
83
            "coalesce(sum(estimated_bytes), 0)"
84
            if "estimated_bytes" in columns
85
            else "0"
86
        )
87
        rows = con.execute(
88
            f"""
89
            SELECT upper(level) AS level, count(*) AS rows,
90
                   coalesce(max(id), 0) AS max_id, {byte_expr} AS bytes
91
            FROM logs
92
            WHERE ts >= ?
93
            GROUP BY upper(level)
94
            ORDER BY rows DESC
95
            """,
96
            (int(time.time()) - seconds,),
97
        ).fetchall()
98
        page_size = int(con.execute("PRAGMA page_size").fetchone()[0])
99
        free_pages = int(con.execute("PRAGMA freelist_count").fetchone()[0])
100

101
        print(f"database: {db} ({db.stat().st_size} bytes)")
102
        wal = Path(str(db) + "-wal")
103
        print(f"wal: {wal.stat().st_size if wal.exists() else 0} bytes")
104
        print(f"trigger: {'OK' if trigger_ok(con) else 'MISSING/OUTDATED'}")
105
        print(f"reclaimable after VACUUM: {page_size * free_pages} bytes")
106
        print(f"recent {seconds}s:")
107
        if not rows:
108
            print("  no rows")
109
        for row in rows:
110
            print(
111
                f"  {row['level']}: rows={row['rows']} "
112
                f"max_id={row['max_id']} bytes={row['bytes']}"
113
            )
114

115
def install(db: Path) -> None:
116
    backup(db)
117
    with connect(db, readonly=False) as con:
118
        assert_schema(con)
119
        for name in KNOWN_TRIGGERS:
120
            con.execute(f"DROP TRIGGER IF EXISTS {name}")
121
        con.execute(TRIGGER_SQL)
122
        con.commit()
123
        if not trigger_ok(con):
124
            fail("trigger verification failed")
125
    print(f"installed: {TRIGGER_NAME}")
126
    print(f"blocked: {', '.join(BLOCKED_LEVELS)}; all other levels are preserved")
127

128
def check(db: Path) -> None:
129
    with connect(db, readonly=True) as con:
130
        assert_schema(con)
131
        if not trigger_ok(con):
132
            fail("trigger missing or outdated")
133
    print(f"OK: {TRIGGER_NAME}")
134

135
def remove(db: Path) -> None:
136
    backup(db)
137
    with connect(db, readonly=False) as con:
138
        assert_schema(con)
139
        for name in KNOWN_TRIGGERS:
140
            con.execute(f"DROP TRIGGER IF EXISTS {name}")
141
        con.commit()
142
    print("removed known guard triggers; default logging restored")
143

144
def main() -> int:
145
    parser = argparse.ArgumentParser(description="Guard Codex SQLite low-level logs")
146
    parser.add_argument("command", choices=("status", "install", "check", "remove"))
147
    parser.add_argument("--db", type=Path, default=default_db())
148
    parser.add_argument("--recent-seconds", type=int, default=300)
149
    args = parser.parse_args()
150
    if args.recent_seconds < 1:
151
        fail("--recent-seconds must be positive")
152
    if args.command == "status":
153
        status(args.db.expanduser(), args.recent_seconds)
154
    elif args.command == "install":
155
        install(args.db.expanduser())
156
    elif args.command == "check":
157
        check(args.db.expanduser())
158
    else:
159
        remove(args.db.expanduser())
160
    return 0
161

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

1
Codex 生成日志
2
      ↓
3
尝试 INSERT 到 logs
4
      ↓
5
SQLite BEFORE INSERT trigger
6
      ├─ TRACE / DEBUG / INFO → RAISE(IGNORE)，取消本行写入
7
      └─ 其他级别            → 正常保存

1
CREATE TRIGGER codex_drop_low_value_logs
2
BEFORE INSERT ON logs
3
WHEN upper(coalesce(NEW.level, '')) IN ('TRACE', 'DEBUG', 'INFO')
4
BEGIN
5
    SELECT RAISE(IGNORE);
6
END;