| 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117 | """Protokoll: jedes Ereignis als eine JSON-Zeile, um Fehler zu finden und
kode zu verbessern.
Je Sitzung eine Datei /var/lib/kode/protokoll/<sitzung>.jsonl, dazu
karte-JJJJ-MM.jsonl fuer alles ohne Sitzung (Waechter, `kode stopp`).
Protokolliert wird alles: Eingaben, jede Anfrage ans Modell mit Messwerten,
jeder Werkzeugaufruf mit Freigabe-Entscheidung und Ergebnis, Kontext-
Kuerzungen, jede Phase des Kartenstarts, Loeschungen, Fehler mit Traceback.
**Achtung:** Es gibt bewusst keine Geheimnis-Sperre (Entscheidung vom
16.10.2026). Was das Modell liest, steht damit auch hier - darum ist der
Ordner nur fuer root lesbar (0700). Einzelne Werte werden bei
MAX_FELD Zeichen abgeschnitten, Dateien nach AUFBEWAHREN_TAGE geloescht.
Schreiben darf nie die Arbeit stoeren: jeder Fehler hier wird geschluckt.
"""
import json
import os
import sys
import threading
import time
import traceback
from pathlib import Path
from typing import Any
from . import konfig
ORDNER = konfig.DATEN / "protokoll"
MAX_FELD = 20000
AUFBEWAHREN_TAGE = 90
VERSION = "1"
_sitzung: str | None = None
_sperre = threading.Lock()
def _datei() -> Path:
if _sitzung:
return ORDNER / f"{_sitzung}.jsonl"
return ORDNER / f"karte-{time.strftime('%Y-%m')}.jsonl"
def _kappen(wert: Any) -> Any:
if isinstance(wert, str) and len(wert) > MAX_FELD:
return wert[:MAX_FELD] + f"…[+{len(wert) - MAX_FELD} Zeichen]"
if isinstance(wert, dict):
return {k: _kappen(v) for k, v in wert.items()}
if isinstance(wert, (list, tuple)):
return [_kappen(v) for v in wert]
return wert
def ereignis(art: str, **daten: Any) -> None:
try:
zeile = {"t": round(time.time(), 3), "zeit": time.strftime("%Y-%m-%d %H:%M:%S"),
"art": art, "pid": os.getpid(), **_kappen(daten)}
text = json.dumps(zeile, ensure_ascii=False, default=str)
with _sperre:
ORDNER.mkdir(parents=True, exist_ok=True)
os.chmod(ORDNER, 0o700)
ziel = _datei()
neu = not ziel.exists()
with open(ziel, "a", encoding="utf-8") as f:
f.write(text + "\n")
if neu:
os.chmod(ziel, 0o600)
except Exception: # noqa: BLE001
pass
def fehler(wo: str, f: BaseException | None = None, **daten: Any) -> None:
f = f or sys.exc_info()[1]
ereignis("fehler", wo=wo, typ=type(f).__name__ if f else None, text=str(f) if f else None,
traceback="".join(traceback.format_exception(f)) if f else None, **daten)
def sitzung_beginnen(kennung: str, **daten: Any) -> None:
global _sitzung
_sitzung = kennung
ereignis("sitzung_start", version=VERSION, python=sys.version.split()[0],
term=os.getenv("TERM"), colorterm=os.getenv("COLORTERM"),
mosh=os.getenv("KODE_UEBER_MOSH") == "1" or _ist_mosh(), tmux=bool(os.getenv("TMUX")), argv=sys.argv[1:], **daten)
_aufraeumen()
alt = sys.excepthook
def haken(typ, wert, tb): # noqa: ANN001
fehler("unbehandelt", wert)
alt(typ, wert, tb)
sys.excepthook = haken
def _ist_mosh() -> bool:
"""Laeuft der Elternprozess-Baum unter mosh-server?"""
try:
pid = os.getppid()
for _ in range(6):
name = Path(f"/proc/{pid}/comm").read_text().strip()
if name.startswith("mosh-server"):
return True
pid = int(Path(f"/proc/{pid}/stat").read_text().split(")")[1].split()[1])
if pid <= 1:
break
except (OSError, ValueError, IndexError):
pass
return False
def _aufraeumen() -> None:
try:
grenze = time.time() - AUFBEWAHREN_TAGE * 86400
for p in ORDNER.glob("*.jsonl"):
if p.stat().st_mtime < grenze:
p.unlink()
except OSError:
pass |