retoor/tai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v1.44.0: extensive debug logging behind --debug

Browse Source 
- Rules first in logging.md: R1-R12 plus a 27-event registry, every
  site validated against the source before implementation.
- DebugLog: file/stdout/stderr sinks, sentence/json formats,
  wipe-on-first-write for roots, append for TAI_CHILD, per-row
  commit, prune to 20000 rows, never raises.
- Vault logs table with six indexes plus FTS triggers, sealed at
  rest, read back through Store.log_query.
- Hooks: turns, completions (backend, model, ms, tokens, texts),
  tool dispatch, approvals, vault mutations, scheduler, serve,
  CAP turns, telegram updates, uncaught exceptions.
- Secrets never logged: redact plus LOG_MASKED_ARGS.
- 13 new tests (1222 + 16 green).
This commit is contained in:
retoor
2026-10-18 21:53:03 +02:00
parent 609e5d8e91
commit 67d4806d82
6 changed files with 975 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -15,6 +15,7 @@
!skills.md
!settings.md
!mcp.md
!logging.md
# Backup files
*.bak
AGENTS.md
+1
View File
@@ -14,6 +14,7 @@
- MCP support (`Mcp*` classes, `mcp_*` tools, `--serve-mcp`) is specified in [mcp.md](mcp.md): read it before touching any `Mcp*` code, and keep its wire, caps, and tool tables in step with the code (`McpDocTests` fails when they drift).
- Skills (`discover_skills`, `SKILL_BLUEPRINTS`, `BUNDLED_SKILLS`, the `skills` install target) are specified in [skills.md](skills.md): read it before touching skill code or embedded skill text, and keep its bundled-skill and adaptation tables in step with the code (`BundledSkillsTests` fails when they drift).
- Settings (`SETTINGS_CATALOG`, the `settings` table, `get_or_create_settings`/`upsert_settings`/`get_all_settings`, `boot_settings`) are specified in [settings.md](settings.md): read it before touching settings code, and keep its catalog table in step with the code (`SettingsTests` fails when they drift).
- Debug logging (`DebugLog`, the module-global `LOG`, `--debug`/`--debug-format`, the `logs` table) is specified in [logging.md](logging.md): read it before touching logging code, and keep its rules, registry, and cap tables in step with the code (`LoggingDocTests` fails when they drift).
## Testing
README.md
+5
View File
@@ -379,6 +379,10 @@ The interactive UI toolkit lives in `tai.py` itself (stdlib-only, no dependencie
Every `TAI_*` knob above that behaves like a setting (model, backends, keys, voices, theme, CLI paths, capability flags) is also overridable from the vault through `get_or_create_settings` / `upsert_settings` / `get_all_settings`, with env always winning. Full contract: [settings.md](settings.md).
## Debug logging
`./tai.py --debug` records every interesting action twice: one line per event to `~/.tai/debug.log` (wiped on each root start, appended by child processes marked with `TAI_CHILD`), and one row per event in the vault `logs` table. `--debug stdout` / `--debug stderr` move the stream; `--debug-format json` switches lines from sentences to JSON (`TAI_DEBUG` / `TAI_DEBUG_FORMAT` do the same from env). Every event carries profile, bot, telegram chat id, a per-boot run id, and fixed meta: LLM completions trap backend, model, milliseconds, tokens in/out/total, and the system/user/assistant texts; tool calls trap name, redacted args, milliseconds, and outcome; turns, approvals, vault writes, scheduler fires, served MCP requests, CAP turns, and telegram updates each log once at their layer. Secret values never reach a log (names only, values redacted, secret-shaped tool args masked). The table is indexed on every filter column plus FTS, sealed at rest, and pruned to the newest 20,000 rows. Full rules and the event registry: [logging.md](logging.md).
## Telegram
./tai.py --install-telegram
@@ -432,5 +436,6 @@ Unit tests live at the top level (`test_tai.py`, `test_seal.py`); integration te
mcp.md MCP client + server (`mcp_*` tools, `--serve-mcp`): wire contract
skills.md skills system and bundled set: provenance, adaptations, install
settings.md vault-backed settings: catalog, precedence, API
logging.md debug logging (`--debug`): rules, event registry, caps
research/ archived design research, registry at research/INDEX.md
AGENTS.md contributor conventions (tests, research rule)
logging.md
+229
View File
@@ -0,0 +1,229 @@
# logging.md
Extensive debug logging for tai, behind `--debug`. One `DebugLog`
owns three outputs: a sink (file, stdout, stderr), the vault `logs`
table, and nothing else. `LoggingDocTests` fails when this file
drifts from the code.
## Flags and sinks
`--debug` takes an optional sink: `--debug` alone means
`--debug file`, `--debug stdout` and `--debug stderr` redirect the
stream. `--debug-format` is `sentence` (default) or `json`.
`TAI_DEBUG` and `TAI_DEBUG_FORMAT` are env fallbacks when the flags
are absent. Without either flag or env var, logging is fully off:
no file, no rows, no overhead beyond one boolean check per hook.
The file sink is `<TAI_HOME>/debug.log`, one line per event,
flushed on every write so a crash loses nothing. File mode also
writes every event to the `logs` table when a store exists; modes
without a store (demo, mplex client, grr client) are file-only.
stdout/stderr sinks still write the table too: the sink only moves
the human stream.
## Wipe rule
The log file is wiped on the first write of a root process, and a
process is root when `TAI_CHILD` is unset. A child tai (currently
the `CapAnswerer` research spawn, which sets `TAI_CHILD=1`) appends
to whatever is there and never wipes. In-process workers (fork,
scheduler) share the open handle, so they neither wipe nor reopen.
Wipe-on-first-write, not wipe-on-boot, so `--version` and other
instant exits never destroy the previous log. Two concurrent roots
on one home are last-wipe-wins, documented, not prevented.
## Formats
Both formats carry the same fields: `ts` (UTC ISO), `level`,
`run` (per-process hex id, correlates everything one boot wrote),
`profile`, `bot`, `chat` (telegram chat id, else empty), then the
dotted event name, the message, and the meta map.
Sentence (default), free text last so the head splits cleanly:
```text
2026-10-18T12:01:02Z INFO run=ab12cd34 profile=work bot=main chat= component.event k=v k="quoted value" :: message
```
Meta values containing whitespace or quotes are double-quoted with
backslash escapes; everything before ` :: ` is whitespace-split
`key=value`, nothing else. JSON is one object per line with sorted
keys:
```json
{"bot": "main", "chat": "", "component": "agent", "event": "turn_end", "level": "INFO", "message": "turn finished", "meta": {"exec_ms": 12340, "steps": 4}, "profile": "work", "run": "ab12cd34", "ts": "2026-10-18T12:01:02Z"}
```
## Vault table
```sql
logs(id INTEGER PRIMARY KEY, profile, bot, chat, ts, run, level, component, event, message, meta_json)
```
Every filter column is indexed: `idx_logs_profile(profile, id)`,
`idx_logs_bot(bot)`, `idx_logs_chat(chat)`, `idx_logs_level(level)`,
`idx_logs_event(component, event)`, `idx_logs_run(run)`. An FTS
trigger indexes each row under `kind='log'` (`trg_logs_ai`, and
`trg_logs_ad` removes pruned rows), so `/search` and `fts_search`
rank log lines like any other vault text. `message` and `meta_json`
are sealed with `tai_enc` whenever the vault seal is enabled, and
unsealed only through `Store.log_query`, which takes the same
filters as the indexes plus `limit` and returns newest first.
The logger opens its own SQLite connection (WAL, 30s busy timeout,
one lock for all threads) instead of sharing `Store.db`, commits every row (so the logger
never holds a write transaction that could stall the main store
connection, and readers never lag), and prunes to `LOG_DB_KEEP`
newest rows every `LOG_DB_PRUNE_EVERY` inserts plus once per
configure. Logging never raises: a failed write is counted in
`dropped` and skipped.
## Caps
| Cap | Value | Meaning |
|---|---|---|
| `LOG_DB_KEEP` | 20000 | newest log rows kept per vault; older pruned with their FTS rows |
| `LOG_DB_PRUNE_EVERY` | 256 | inserts between prune checks |
| `LOG_MSG_MAX` | 2000 | message truncation, chars |
| `LOG_TEXT_MAX` | 4000 | each trapped text (system/user/assistant) in DB meta, chars |
| `LOG_PREVIEW` | 300 | each trapped text in the file sink, chars |
| `LOG_META_MAX` | 16384 | `meta_json` truncation, bytes |
## The rules
R1. One event per interesting action, at exactly one site. No
action logs twice: `Tools.dispatch` covers all 61 tools, so no
`run_*` method logs; the bridge, the scheduler watcher, and the
serve dispatcher each log once for their layer.
R2. Level follows consequence, not volume. `INFO` for actions
taken, `WARNING` for retries and refused auth, `ERROR` for failed
operations and uncaught exceptions, `DEBUG` for durability
mutations (record/secret/schedule writes). There is no TRACE:
anything worth trapping is worth INFO.
R3. Every event carries the execution context it ran in: profile,
bot, chat (telegram only, else empty), and the process run id.
Callers pass what they know; the logger merges its sticky
defaults under explicit arguments.
R4. LLM completions trap the full accounting tuple: backend,
model, `exec_ms`, `tokens_in`, `tokens_out`, `tokens_total`,
`cost_usd`, message char counts, and the system, user, and
assistant texts (full to `LOG_TEXT_MAX` in the DB, previewed to
`LOG_PREVIEW` in the file). Tokens are estimated at 4 chars each
unless a backend reports real usage (none do today);
`cost_usd` is null until a backend reports cost.
R5. Timed operations always log `exec_ms` as an integer. The
measurement wraps the operation, never the logging.
R6. Secret values, passphrases, keys, tokens, and header values
never reach a log. Names are fine (`secret_saved name=wifi`).
Every trapped text passes through `Store.redact` for its profile
before truncation, so a known value lands as `[redacted:NAME]`;
tool arguments whose key names a secret (`LOG_MASKED_ARGS`) are
masked to `***` before logging, which also covers values that are
not stored yet and therefore unknown to `redact`.
R7. Errors log where they are decided, with `short_error` text:
backend failure at the raise site, tool failure at dispatch, auth
refusal at the check, uncaught exceptions via `sys.excepthook`.
A catch that only re-raises or translates never logs.
R8. Hot paths never log: TUI render, input, and compositor paths,
token estimators, `redact` itself, FTS internals, health records,
and everything inside `DebugLog`. A hook in a per-keystroke or
per-frame path is a bug, not coverage.
R9. The logger never raises, never blocks on the network, never
touches a tool, and never calls back into logged code. A logging
failure is a dropped event, counted, not an error event.
R10. Child processes append, roots wipe (see Wipe rule). Log volume
is bounded by the caps above, never by sampling: pruning deletes
oldest first, deterministically.
R11. Sentence format keeps free text after ` :: ` and quotes any
meta value containing whitespace. JSON format sorts keys. Both are
line-delimited and greppable without a parser.
R12. New events extend the registry below with a fixed level, a
single site, and a fixed meta vocabulary (`exec_ms`, `tokens_in`,
`tokens_out`, `tokens_total`, `cost_usd`, `outcome`, `error`,
`steps`, `*_chars`, `*_text`). Synonyms are rejected in review.
## Event registry
| Event | Level | Site | Meta |
|---|---|---|---|
| `app.boot` | INFO | `boot()` end | `version`, `home`, `mode`, `profile` |
| `app.exception` | ERROR | `sys.excepthook` | `error` |
| `agent.turn_start` | INFO | `run_turn` wrapper | `goal`, `capture`, `max_steps`, `depth`, `routed` |
| `agent.turn_end` | INFO | `run_turn` wrapper | `exec_ms`, `steps`, `tools_used`, `outcome`, `reply_chars` |
| `agent.verified` | INFO | `ensure_verified` end | `success`, `failures` |
| `agent.compact` | INFO | `compact_messages` end | `before_tokens`, `after_tokens`, `kept_turns` |
| `llm.completion` | INFO | `try_candidate` success | `backend`, `model`, `exec_ms`, `tokens_in`, `tokens_out`, `tokens_total`, `cost_usd`, `history_msgs`, `system_chars`, `user_chars`, `assistant_chars`, `system_text`, `user_text`, `assistant_text`, `tool_calls`, `retried_without_tools` |
| `llm.failed` | WARNING | `try_candidate` failure | `backend`, `model`, `error`, `transient` |
| `llm.round_retry` | WARNING | `complete()` before sleep | `round`, `pause_s`, `candidates`, `errors` |
| `llm.fatal` | ERROR | `complete()` wrapper on `BackendError` | `error` |
| `tool.call` | INFO | `Tools.dispatch` | `name`, `args`, `exec_ms`, `outcome`, `error`, `result_chars` |
| `approval.decision` | INFO | `ask_approval` wrapper | `command`, `decision`, `mode` |
| `vault.record_saved` | DEBUG | `add_record` end | `id`, `kind`, `size` |
| `vault.record_deleted` | DEBUG | `delete_record` end | `id` |
| `vault.secret_saved` | DEBUG | `save_secret` end | `name` |
| `vault.secret_deleted` | DEBUG | `delete_secret` end | `name` |
| `vault.schedule_saved` | DEBUG | `add_schedule` end | `id`, `name` |
| `vault.schedule_removed` | DEBUG | `remove_schedule` end | `id` |
| `vault.seal_rotated` | INFO | `rotate_seal` end | `home` |
| `vault.schema_upgraded` | INFO | `boot()` after print | `notes` |
| `sched.fired` | INFO | `scheduler_tick` after spawn | `id`, `profile`, `every_sec` |
| `sched.finished` | INFO | `watch_schedule` end | `id`, `status`, `result_chars` |
| `serve.request` | INFO | `dispatch_message` | `method`, `tool`, `session`, `exec_ms`, `outcome` |
| `serve.denied` | WARNING | `do_POST` auth check | `path` |
| `cap.run_start` | INFO | `cap_run_main` head | `sessions` |
| `cap.turn` | INFO | `CapFleet.ask` | `target`, `payload_chars`, `exec_ms`, `reply_chars`, `outcome` |
| `telegram.update` | INFO | `handle_telegram_update` wrapper | `chat`, `text_chars`, `reply_chars`, `voice`, `outcome` |
Outcomes are from a fixed set: `ok`, `error`, `denied`,
`unknown_tool`, `bad_args`, `timeout`. `goal`, `args`, `command`,
and all `*_text` values are redacted then truncated; `*_chars`
always reports the pre-truncation length.
## Validation against the source
Every registry site was checked before implementation: `boot`,
`run_turn`, `ensure_verified`, `compact_messages`,
`try_candidate`, `complete`, `Tools.dispatch`, `ask_approval`,
`add_record`, `delete_record`, `save_secret`, `delete_secret`,
`add_schedule`, `remove_schedule`, `rotate_seal`,
`scheduler_tick`, `watch_schedule`, `dispatch_message`, `do_POST`,
`cap_run_main`, `CapFleet.ask`, and `handle_telegram_update` all
exist with the shapes the hooks assume. Adjusted during review:
serve logging moved from `do_POST` (seven early returns) into
`dispatch_message` plus one auth line; scheduler completion moved
from `poll_agent` (shared, would double-log forks) into
`watch_schedule`; uncaught exceptions moved from a `main()` wrap
into `sys.excepthook`; `run_turn`, `complete`, and `handle_telegram_update` became thin
wrappers over renamed inners so their many exits log once;
`ask_approval` instead logs through an inline `decide()` closure,
because `TuiChatTests` calls it unbound on a stub that a rename
would break.
## Code map
`LOG_*` caps pin the bounds; `DebugLog` owns sink, format, run
id, wipe-on-first-write, the DB connection, prune, and redact;
module-global `LOG` is the one logger, inert until `configure()`;
`TAI_CHILD` selects append; `LOG_MASKED_ARGS` masks secret-shaped
tool arguments; `Store.log_query` reads rows back seal-aware.
`CapAnswerer.research` marks its child.
## Test hooks
`LoggingDocTests` asserts this file names every `LOG_*` cap, every
registry event, the `logs` table with its columns and indexes, the
`trg_logs_*` triggers, both flags, both formats, and `TAI_CHILD`,
with no em dash. `LoggerTests` covers formats, wipe/append,
redaction, prune, seal, `log_query`, the never-raise rule, and
representative hooks (dispatch, completion, approval, turn).
tai.py
+533 -22
View File
@@ -105,7 +105,7 @@ def _fromisoformat(text):
return parsed.replace(tzinfo=tzinfo)
raise ValueError("invalid datetime, use ISO like 2026-10-08T09:00")
VERSION = "1.43.1"
VERSION = "1.44.0"
WORKER_STEPS = 12
CREATE_SKILL_STEPS = 40
FORK_MAX_DEPTH = 2
@@ -2426,6 +2426,7 @@ def rotate_seal(config, old_seal, new_passphrase):
for path, plain in files.items():
with open(path, "w", encoding="utf-8") as handle:
handle.write(fresh.lock(plain))
LOG.event("INFO", "vault", "seal_rotated", "vault seal rotated", meta={"home": config.home})
return fresh
@@ -5350,6 +5351,226 @@ def open_mem_events():
return mem
LOG_DB_KEEP = 20000
LOG_DB_PRUNE_EVERY = 256
LOG_MSG_MAX = 2000
LOG_TEXT_MAX = 4000
LOG_PREVIEW = 300
LOG_META_MAX = 16384
LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR")
LOG_MASKED_ARGS = frozenset(("value", "password", "passphrase", "token", "secret", "key", "auth", "authorization"))
class DebugLog(object):
"""Extensive debug logging: one line per event to a sink, plus the vault logs table. Full rules: logging.md."""
def __init__(self):
self.enabled = False
self.sink = "file"
self.format = "sentence"
self.home = ""
self.handle = None
self.wiped = False
self.run = uuid.uuid4().hex[:8]
self.store = None
self.db = None
self.db_count = 0
self.dropped = 0
self.lock = threading.RLock()
self.context = {}
self.old_excepthook = None
def configure(self, sink="file", fmt="sentence", home="", store=None):
self.reset()
self.sink = sink if sink in ("file", "stdout", "stderr") else "file"
self.format = fmt if fmt in ("sentence", "json") else "sentence"
self.home = home
self.enabled = True
if store is not None:
self.attach_store(store)
self.old_excepthook = sys.excepthook
sys.excepthook = self._excepthook
self.prune()
def reset(self):
try:
self.close()
except Exception:
pass
if self.old_excepthook is not None:
try:
sys.excepthook = self.old_excepthook
except Exception:
pass
self.enabled = False
self.sink = "file"
self.format = "sentence"
self.home = ""
self.handle = None
self.wiped = False
self.run = uuid.uuid4().hex[:8]
self.store = None
self.db = None
self.db_count = 0
self.dropped = 0
self.context = {}
self.old_excepthook = None
def attach_store(self, store):
self.store = store
try:
self.db = sqlite3.connect(store.config.db_path, timeout=30, check_same_thread=False)
self.db.execute("PRAGMA journal_mode=WAL")
self.db.execute("PRAGMA busy_timeout=30000")
self.db.create_function("tai_enc", 1, store.seal.lock)
except sqlite3.Error:
self.db = None
def close(self):
handle, self.handle = self.handle, None
if handle is not None and self.sink == "file":
try:
handle.close()
except (OSError, ValueError):
pass
db, self.db = self.db, None
if db is not None:
try:
db.commit()
db.close()
except sqlite3.Error:
pass
def _open_handle(self):
if self.sink == "stdout":
return sys.stdout
if self.sink == "stderr":
return sys.stderr
os.makedirs(self.home, exist_ok=True)
if not self.wiped and not os.environ.get("TAI_CHILD"):
mode = "w"
else:
mode = "a"
self.wiped = True
return open(os.path.join(self.home, "debug.log"), mode, encoding="utf-8")
@staticmethod
def flat(text):
return " ".join(str(text or "").split())
def scrub(self, text, profile):
if not isinstance(text, str) or not text:
return text
if self.store is None:
return text
try:
return self.store.redact(text, profile or None)
except Exception:
return text
def cut_meta(self, meta, profile, limit):
out = {}
for key in sorted((meta or {})):
value = (meta or {})[key]
if isinstance(value, str):
clean = self.scrub(value, profile)
if len(clean) > limit:
clean = clean[:limit] + "..."
out[str(key)] = self.flat(clean)
elif isinstance(value, (dict, list)):
try:
blob = json.dumps(value, sort_keys=True)[:limit]
except (TypeError, ValueError):
blob = short_error(value)
out[str(key)] = self.flat(self.scrub(blob, profile))
else:
out[str(key)] = value
return out
@staticmethod
def quote(value):
if not isinstance(value, str):
try:
return json.dumps(value, sort_keys=True)
except (TypeError, ValueError):
return short_error(value)
if value == "" or any(char.isspace() or char in "\"\\" for char in value):
return "\"" + value.replace("\\", "\\\\").replace("\"", "\\\"") + "\""
return value
def sentence(self, ts, level, run, profile, bot, chat, component, event, message, meta):
parts = [ts, level, "run=%s" % run, "profile=%s" % self.quote(profile), "bot=%s" % self.quote(bot),
"chat=%s" % self.quote(chat), "%s.%s" % (component, event)]
parts += ["%s=%s" % (key, self.quote(meta[key])) for key in sorted(meta)]
return " ".join(parts) + " :: " + message
@staticmethod
def json_line(ts, level, run, profile, bot, chat, component, event, message, meta):
return json.dumps({"ts": ts, "level": level, "run": run, "profile": profile, "bot": bot, "chat": chat,
"component": component, "event": event, "message": message, "meta": meta}, sort_keys=True)
def event(self, level, component, event, message, profile="", bot="", chat="", meta=None):
if not self.enabled:
return
try:
with self.lock:
profile = profile or self.context.get("profile", "")
bot = bot or self.context.get("bot", "")
chat = chat or self.context.get("chat", "")
clean = self.flat(self.scrub(message, profile))[:LOG_MSG_MAX]
db_meta = self.cut_meta(meta, profile, LOG_TEXT_MAX)
show_meta = self.cut_meta(meta, profile, LOG_PREVIEW)
ts = datetime.now(timezone.utc).isoformat()
if self.format == "json":
line = self.json_line(ts, level, self.run, profile, bot, chat, component, event, clean, show_meta)
else:
line = self.sentence(ts, level, self.run, profile, bot, chat, component, event, clean, show_meta)
if self.handle is None:
self.handle = self._open_handle()
self.handle.write(line + "\n")
self.handle.flush()
if self.db is not None:
blob = json.dumps(db_meta, sort_keys=True)[:LOG_META_MAX]
if self.store is not None and self.store.seal.enabled:
self.db.execute("INSERT INTO logs (profile, bot, chat, ts, run, level, component, event, message, meta_json) VALUES (?, ?, ?, ?, ?, ?, ?, ?, tai_enc(?), tai_enc(?))",
(profile, bot, chat, ts, self.run, level, component, event, clean, blob))
else:
self.db.execute("INSERT INTO logs (profile, bot, chat, ts, run, level, component, event, message, meta_json) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(profile, bot, chat, ts, self.run, level, component, event, clean, blob))
self.db_count += 1
self.db.commit()
if self.db_count % LOG_DB_PRUNE_EVERY == 0:
self.prune()
except Exception:
self.dropped += 1
def prune(self):
if self.db is None:
return
try:
with self.lock:
self.db.execute("DELETE FROM logs WHERE id NOT IN (SELECT id FROM logs ORDER BY id DESC LIMIT ?)", (LOG_DB_KEEP,))
self.db.commit()
except sqlite3.Error:
pass
def _excepthook(self, typ, value, tb):
try:
if not issubclass(typ, (KeyboardInterrupt, SystemExit, GeneratorExit)):
self.event("ERROR", "app", "exception", "uncaught %s" % short_error(value),
meta={"error": "%s: %s" % (typ.__name__, short_error(value))})
except Exception:
pass
if self.old_excepthook is not None:
try:
self.old_excepthook(typ, value, tb)
except Exception:
pass
LOG = DebugLog()
def setup_fts(db):
try:
exists = db.execute("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'fts_docs'").fetchone()
@@ -5384,6 +5605,12 @@ def setup_fts(db):
db.execute("""CREATE TRIGGER IF NOT EXISTS trg_chunks_ad AFTER DELETE ON chunks BEGIN
DELETE FROM fts_docs WHERE item = 'chunk:' || old.id AND kind = 'chunk';
END""")
db.execute("""CREATE TRIGGER IF NOT EXISTS trg_logs_ai AFTER INSERT ON logs BEGIN
INSERT INTO fts_docs(item, kind, sub, title, body, profile) VALUES('log:' || new.id, 'log', new.component || '/' || new.event, new.ts || ' ' || new.level || ' ' || new.component || '/' || new.event, new.message, new.profile);
END""")
db.execute("""CREATE TRIGGER IF NOT EXISTS trg_logs_ad AFTER DELETE ON logs BEGIN
DELETE FROM fts_docs WHERE item = 'log:' || old.id AND kind = 'log';
END""")
if db.execute("SELECT COUNT(*) FROM fts_docs").fetchone()[0] == 0:
db.execute("INSERT INTO fts_docs(item, kind, sub, title, body, profile) SELECT id, 'record', kind, title, content, profile FROM records")
db.execute("INSERT INTO fts_docs(item, kind, sub, title, body, profile) SELECT id, 'event', role || '/' || kind, ts || ' ' || role || '/' || kind, text, profile FROM events")
@@ -5441,6 +5668,7 @@ SCHEMA = (
("mcp_server_profiles", ("server TEXT", "profile TEXT", "enabled INTEGER"), ("server", "profile")),
("mcp_tool_pins", ("server TEXT", "tool TEXT", "hash TEXT", "updated TEXT"), ("server", "tool")),
("mcp_server_tools", ("server TEXT", "tool TEXT", "description TEXT", "schema_json TEXT", "updated TEXT"), ("server", "tool")),
("logs", ("id INTEGER PRIMARY KEY", "profile TEXT", "bot TEXT", "chat TEXT", "ts TEXT", "run TEXT", "level TEXT", "component TEXT", "event TEXT", "message TEXT", "meta_json TEXT"), ()),
)
SCHEMA_INDEXES = (
@@ -5457,6 +5685,12 @@ SCHEMA_INDEXES = (
("idx_chunks_profile", "chunks(profile)"),
("idx_input_history_profile", "input_history(profile, id)"),
("idx_shell_sessions_parent", "shell_sessions(parent)"),
("idx_logs_profile", "logs(profile, id)"),
("idx_logs_bot", "logs(bot)"),
("idx_logs_chat", "logs(chat)"),
("idx_logs_level", "logs(level)"),
("idx_logs_event", "logs(component, event)"),
("idx_logs_run", "logs(run)"),
)
@@ -5506,6 +5740,8 @@ class Store:
self.db.execute("UPDATE secrets SET meta = tai_enc(COALESCE(meta, '{}')) WHERE meta IS NULL OR meta NOT LIKE 'tai1$%'")
self.db.execute("UPDATE schedules SET prompt = tai_enc(prompt) WHERE prompt NOT LIKE 'tai1$%'")
self.db.execute("UPDATE input_history SET text = tai_enc(text) WHERE text NOT LIKE 'tai1$%'")
self.db.execute("UPDATE logs SET message = tai_enc(message) WHERE message NOT LIKE 'tai1$%'")
self.db.execute("UPDATE logs SET meta_json = tai_enc(meta_json) WHERE meta_json NOT LIKE 'tai1$%'")
sealed_keys = [key for key, spec in SETTINGS_CATALOG.items() if spec["encrypted"]]
if sealed_keys:
self.db.execute(
@@ -5645,6 +5881,36 @@ class Store:
except sqlite3.Error:
pass
def log_query(self, level=None, component=None, event=None, profile=None, bot=None, chat=None, run=None, limit=100):
"""Read debug log rows newest first, seal-aware. Every filter hits an index."""
message_col = "tai_dec(message)" if self.seal.enabled else "message"
meta_col = "tai_dec(meta_json)" if self.seal.enabled else "meta_json"
query = "SELECT id, profile, bot, chat, ts, run, level, component, event, %s, %s FROM logs" % (message_col, meta_col)
clauses, params = [], []
for column, value in (("level", level), ("component", component), ("event", event), ("profile", profile),
("bot", bot), ("chat", chat), ("run", run)):
if value:
clauses.append(column + " = ?")
params.append(value)
if clauses:
query += " WHERE " + " AND ".join(clauses)
query += " ORDER BY id DESC LIMIT ?"
params.append(max(1, min(limit or 100, 1000)))
try:
rows = self.db.execute(query, params).fetchall()
except sqlite3.Error:
return []
found = []
for row in rows:
try:
meta = json.loads(row[10] or "{}")
except ValueError:
meta = {}
found.append({"id": row[0], "profile": row[1], "bot": row[2], "chat": row[3], "ts": row[4], "run": row[5],
"level": row[6], "component": row[7], "event": row[8], "message": row[9],
"meta": meta if isinstance(meta, dict) else {}})
return found
def fts_search(self, query, kinds=(), profile=None, limit=10):
profile = self._scope(profile)
if not self.fts_ok:
@@ -5661,7 +5927,7 @@ class Store:
sql += " AND kind IN (%s)" % marks
params_tail += list(kinds)
if self.seal.enabled:
sql += " AND kind != 'event'"
sql += " AND kind != 'event' AND kind != 'log'"
sql += " ORDER BY %s LIMIT ?" % rank
mem_rank = fts_rank("mem_events")
mem_sql = "SELECT item, kind, sub, title, snippet(mem_events, 4, '[', ']', '...', 12), %s FROM mem_events WHERE mem_events MATCH ? AND profile = ? ORDER BY %s LIMIT ?" % (mem_rank, mem_rank)
@@ -5745,6 +6011,7 @@ class Store:
else:
self.set_item_tags("secret:" + name, ["secret"] + list(tags), profile)
self.db.commit()
LOG.event("DEBUG", "vault", "secret_saved", "secret %s saved" % name, profile=profile, meta={"name": name})
def secret_meta(self, name, profile=None):
profile = self._scope(profile)
@@ -5905,6 +6172,8 @@ class Store:
for other, shared in self.tag_neighbors(record_id, None, profile):
self.db.execute("INSERT OR IGNORE INTO edges(src, dst, relation, profile, created) VALUES(?, ?, ?, ?, ?)", (record_id, other, "shares-" + shared, profile, stamp))
self.db.commit()
LOG.event("DEBUG", "vault", "record_saved", "record %s saved (%s, %d bytes)" % (record_id, kind, size),
profile=profile, meta={"id": record_id, "kind": kind, "size": size})
return record_id
def add_chunks(self, parent_kind, parent_id, texts, profile=None):
@@ -6007,6 +6276,8 @@ class Store:
self.db.execute("DELETE FROM edges WHERE profile = ? AND (src = ? OR dst = ?)", (profile, record_id, record_id))
self.db.execute("DELETE FROM chunks WHERE parent_kind = 'record' AND parent_id = ? AND profile = ?", (record_id, profile))
self.db.commit()
LOG.event("DEBUG", "vault", "record_deleted", "record %s deleted" % record_id,
profile=profile, meta={"id": record_id})
return True
self.db.commit()
return False
@@ -6145,6 +6416,7 @@ class Store:
self.db.execute("DELETE FROM tags WHERE item = ? AND profile = ?", (item, profile))
self.db.execute("DELETE FROM edges WHERE profile = ? AND (src = ? OR dst = ?)", (profile, item, item))
self.db.commit()
LOG.event("DEBUG", "vault", "secret_deleted", "secret %s deleted" % name, profile=profile, meta={"name": name})
return True
self.db.commit()
return False
@@ -6180,6 +6452,8 @@ class Store:
else:
self.set_item_tags(item, ["schedule"] + list(tags), profile)
self.db.commit()
LOG.event("DEBUG", "vault", "schedule_saved", "schedule #%s saved (%s)" % (done.lastrowid, name),
profile=profile, meta={"id": done.lastrowid, "name": name})
return done.lastrowid
def remove_schedule(self, row_id, profile=None):
@@ -6190,6 +6464,8 @@ class Store:
self.db.execute("DELETE FROM tags WHERE item = ? AND profile = ?", (item, profile))
self.db.execute("DELETE FROM edges WHERE profile = ? AND (src = ? OR dst = ?)", (profile, item, item))
self.db.commit()
LOG.event("DEBUG", "vault", "schedule_removed", "schedule #%s removed" % row_id,
profile=profile, meta={"id": row_id})
return True
self.db.commit()
return False
@@ -7951,6 +8227,15 @@ class ChatClient:
return text[:width]
def complete(self, messages, tools=None, stream_sink=None, ephemeral=False, deadline=None):
try:
return self._complete_inner(messages, tools, stream_sink, ephemeral, deadline)
except BackendError as exc:
profile, bot = self.scope
LOG.event("ERROR", "llm", "fatal", "all backends failed: %s" % short_error(exc),
profile=profile, bot=bot, meta={"error": short_error(exc)})
raise
def _complete_inner(self, messages, tools=None, stream_sink=None, ephemeral=False, deadline=None):
messages = clean_history(messages)
payload = {"model": self.config.model, "messages": messages}
if tools:
@@ -7978,6 +8263,12 @@ class ChatClient:
if give_up is not None and time.time() + pause > give_up:
raise BackendError("all backends failed after %d rounds (%s)" % (round_no, self.summarize_errors(errors, shown=6, width=900)))
providers = len(set((cand["provider"], cand["chat_url"]) for cand in cands))
profile, bot = self.scope
LOG.event("WARNING", "llm", "round_retry",
"all %d candidates failed (round %d), retrying in %ds" % (len(cands), round_no, pause),
profile=profile, bot=bot,
meta={"round": round_no, "pause_s": pause, "candidates": len(cands),
"errors": self.summarize_errors(errors)})
if sys.stderr.isatty():
sys.stderr.write("\n")
print(paint("all %d candidate%s across %d backend%s failed (round %d), retrying in %ds: %s" % (len(cands), "" if len(cands) == 1 else "s", providers, "" if providers == 1 else "s", round_no, pause, self.summarize_errors(errors)), Ansi.DIM), file=sys.stderr)
@@ -8072,21 +8363,62 @@ class ChatClient:
if not isinstance(reply, dict):
raise BackendError("backend returned no reply")
self.health.record(label, model, True, (time.time() - started) * 1000, max(1, len(reply.get("content") or "") // 4) / max(time.time() - started, 0.001))
if LOG.enabled:
self.log_completion(cand, messages, reply, use_tools, (time.time() - started) * 1000, pos > 0)
return reply
except BackendError as exc:
if pos == 0 and exc.status == 400 and len(tries) > 1:
continue
self.health.record(label, model, False)
self.round_transient = self.round_transient or transient_failure(exc)
transient = transient_failure(exc)
self.round_transient = self.round_transient or transient
errors.append("%s/%s: %s" % (label, model, exc))
profile, bot = self.scope
LOG.event("WARNING", "llm", "failed", "%s/%s failed: %s" % (label, model, short_error(exc)),
profile=profile, bot=bot,
meta={"backend": label, "model": model, "error": short_error(exc), "transient": transient})
return None
except Exception as exc:
self.health.record(label, model, False)
self.round_transient = True
errors.append("%s/%s: %s" % (label, model, short_error(exc)))
profile, bot = self.scope
LOG.event("WARNING", "llm", "failed", "%s/%s failed: %s" % (label, model, short_error(exc)),
profile=profile, bot=bot,
meta={"backend": label, "model": model, "error": short_error(exc), "transient": True})
return None
return None
def log_completion(self, cand, messages, reply, use_tools, elapsed_ms, retried):
profile, bot = self.scope
def text_of(item):
content = item.get("content")
if isinstance(content, str):
return content
if content is None:
return ""
try:
return json.dumps(content, sort_keys=True)
except (TypeError, ValueError):
return short_error(content)
system_text = next((text_of(item) for item in messages if item.get("role") == "system"), "")
user_text = next((text_of(item) for item in reversed(messages) if item.get("role") == "user"), "")
assistant_text = reply.get("content") or ""
tokens_in = conversation_tokens(messages, use_tools or [])
tokens_out = estimate_tokens(assistant_text) + estimate_tokens(json.dumps(reply.get("tool_calls") or []))
calls = reply.get("tool_calls") or []
LOG.event("INFO", "llm", "completion",
"%s/%s answered in %dms (%d tokens in, %d out)" % (cand["provider"], cand["model"], int(elapsed_ms), tokens_in, tokens_out),
profile=profile, bot=bot,
meta={"backend": cand["provider"], "model": cand["model"], "exec_ms": int(elapsed_ms),
"tokens_in": tokens_in, "tokens_out": tokens_out, "tokens_total": tokens_in + tokens_out,
"cost_usd": None, "history_msgs": len(messages), "system_chars": len(system_text),
"user_chars": len(user_text), "assistant_chars": len(assistant_text),
"system_text": system_text, "user_text": user_text, "assistant_text": assistant_text,
"tool_calls": len(calls), "retried_without_tools": retried})
def single_shot(self, url, key, payload, label, extra=None):
body = json.dumps(payload).encode("utf-8")
request = urllib.request.Request(url, data=body, headers=self.headers(key, extra), method="POST")
@@ -17075,9 +17407,11 @@ class Tools:
def dispatch(self, name, raw_arguments):
self.calls += 1
started = time.time()
try:
args = raw_arguments if isinstance(raw_arguments, dict) else json.loads(raw_arguments or "{}")
except (TypeError, ValueError):
self.log_call(name, raw_arguments, started, "bad_args", "error: arguments are not valid JSON")
return "error: arguments are not valid JSON"
handler = self.handlers.get(name)
if handler is None and name.startswith(MCP_TOOL_PREFIX):
@@ -17086,17 +17420,45 @@ class Tools:
server, tool = resolved
handler = lambda call_args: McpBridge(self.app).call(server, tool, call_args, self.active_profile)
if handler is None:
self.log_call(name, args, started, "unknown_tool", "error: unknown tool " + name)
return "error: unknown tool " + name
try:
result = handler(args)
except Denied:
self.log_call(name, args, started, "denied", "denied")
raise
except Exception as exc:
self.log_call(name, args, started, "error", "error: " + short_error(exc))
return "error: " + short_error(exc)
if self.app.store is not None:
return self.app.store.redact(result, self.active_profile)
result = self.app.store.redact(result, self.active_profile)
self.log_call(name, args, started, "ok", result)
return result
def log_call(self, name, args, started, outcome, result):
if not LOG.enabled:
return
exec_ms = int((time.time() - started) * 1000)
if isinstance(args, dict):
shown = dict((key, "***" if str(key).lower() in LOG_MASKED_ARGS else args[key]) for key in args)
else:
shown = args if isinstance(args, str) else short_error(args)
text = result if isinstance(result, str) else ""
if outcome == "ok":
message = "tool %s finished in %dms" % (name, exec_ms)
elif outcome == "denied":
message = "tool %s denied in %dms" % (name, exec_ms)
elif outcome == "unknown_tool":
message = "tool %s unknown" % name
elif outcome == "bad_args":
message = "tool %s got bad arguments" % name
else:
message = "tool %s failed in %dms: %s" % (name, exec_ms, short_error(result))
meta = {"name": name, "args": shown, "exec_ms": exec_ms, "outcome": outcome, "result_chars": len(text)}
if outcome == "error":
meta["error"] = short_error(result)
LOG.event("INFO", "tool", "call", message, profile=self.active_profile, bot=getattr(self.app, "bot", ""), meta=meta)
def secret_grant(self, kind, key, names):
token = (kind, key, tuple(sorted(names)))
if token in self.secret_grants:
@@ -18703,12 +19065,18 @@ def compact_messages(messages, chat, keep=KEEP_TURNS):
{"role": "system", "content": COMPACT_SYSTEM},
{"role": "user", "content": "Compress this history:\n" + "\n".join(digest)},
]
before = conversation_tokens(messages)
summary = chat.complete(request, ephemeral=True)["content"].strip()
rebuilt = [
system,
{"role": "user", "content": "Previous session summary:\n" + summary},
{"role": "assistant", "content": "Understood. Continuing with summarized context."},
] + recent
after = conversation_tokens(rebuilt)
profile, bot = getattr(chat, "scope", ("", ""))
LOG.event("INFO", "agent", "compact", "compacted context %d to %d tokens" % (before, after),
profile=profile, bot=bot,
meta={"before_tokens": before, "after_tokens": after, "kept_turns": len(recent)})
return rebuilt, True
@@ -19260,10 +19628,25 @@ class Agent:
self.chat.reset_muse()
def ask_approval(self, command, guidance=True):
if self.yolo or self.auto or not self.persist:
mode = "auto"
elif getattr(self, "prompter", None) is not None:
mode = "prompter"
elif sys.stdin.isatty():
mode = "tty"
else:
mode = "headless"
def decide(value):
LOG.event("INFO", "approval", "decision", "approval %s" % ("granted" if value else "denied"),
profile=getattr(self, "profile", ""), bot=getattr(self, "bot", ""),
meta={"command": (command or "")[:300], "decision": value, "mode": mode})
return value
if self.yolo or self.auto:
return True
return decide(True)
if not self.persist:
return self.config.auto_approve
return decide(self.config.auto_approve)
prompter = getattr(self, "prompter", None)
if prompter is not None:
choice = prompter("approve", command, ["Allow once", "Always allow (approvals off)", "Deny"])
@@ -19271,20 +19654,20 @@ class Agent:
self.user_stepped_in = True
self.yolo = True
self.config.auto_approve = True
return True
return decide(True)
if choice == "Allow once":
self.user_stepped_in = True
return True
return False
return decide(True)
return decide(False)
if not sys.stdin.isatty():
return False
return decide(False)
print(paint("run: %s" % command, Ansi.YELLOW))
decision = "unknown"
for _retry in range(APPROVAL_MAX_RETRIES):
try:
answer = input(paint("allow? [y]once [Y]always [n]o ", Ansi.YELLOW)).strip()
except (EOFError, KeyboardInterrupt):
return False
return decide(False)
decision = Approval.parse_allow(answer)
if decision != "unknown":
break
@@ -19294,15 +19677,16 @@ class Agent:
self.yolo = True
self.config.auto_approve = True
print(paint("yolo mode on, approvals off for this session", Ansi.YELLOW))
return True
return decide(True)
if decision == "once":
return True
return decide(True)
if not guidance or self.auto:
return False
return decide(False)
try:
typed = input(paint("denied. What should I do instead? (empty aborts this turn) ", Ansi.YELLOW)).strip()
except (EOFError, KeyboardInterrupt):
typed = ""
decide(False)
raise Denied(typed)
def show_call(self, call):
@@ -19484,6 +19868,25 @@ class Agent:
pass
def run_turn(self, text, capture=False, max_steps=MAX_STEPS, _routed=False):
LOG.event("INFO", "agent", "turn_start", "turn started", profile=self.profile, bot=self.bot,
meta={"goal": (text or "")[:500], "capture": capture, "max_steps": max_steps,
"depth": self.depth, "routed": _routed})
started = time.time()
try:
result = self._run_turn_inner(text, capture, max_steps, _routed)
except Exception as exc:
LOG.event("INFO", "agent", "turn_end", "turn failed: %s" % short_error(exc),
profile=self.profile, bot=self.bot,
meta={"exec_ms": int((time.time() - started) * 1000), "steps": getattr(self, "_turn_steps", 0),
"tools_used": getattr(self, "_turn_tools", 0), "outcome": "error", "reply_chars": 0})
raise
LOG.event("INFO", "agent", "turn_end", "turn finished", profile=self.profile, bot=self.bot,
meta={"exec_ms": int((time.time() - started) * 1000), "steps": getattr(self, "_turn_steps", 0),
"tools_used": getattr(self, "_turn_tools", 0), "outcome": "ok",
"reply_chars": len(result or "")})
return result
def _run_turn_inner(self, text, capture=False, max_steps=MAX_STEPS, _routed=False):
self.apply_system()
if not _routed:
text, result = self.apply_mentions(text, capture, max_steps)
@@ -19651,7 +20054,11 @@ class Agent:
"""Verify a finished one-shot turn against its goal, running
follow-up turns until the verdict passes or VERIFY_MAX_ROUNDS is
spent. Returns only whether the goal was delivered."""
return self.verify_loop(goal, final)[1]
final_text, success = self.verify_loop(goal, final)
failures = 0 if success else getattr(self, "_last_verify_failures", 1)
LOG.event("INFO", "agent", "verified", "goal %s" % ("delivered" if success else "not delivered"),
profile=self.profile, bot=self.bot, meta={"success": success, "failures": failures})
return success
def verify_loop(self, goal, final):
goal = " ".join(str(goal or "").split())[:VERIFY_MAX_GOAL]
@@ -19660,6 +20067,7 @@ class Agent:
return final, False
for _attempt in range(1 + VERIFY_MAX_ROUNDS):
verdict = self.verify_goal(goal, final)
self._last_verify_failures = len(verdict.failures)
self.show_verdict(verdict, goal)
self.store.log_event(self.profile, "system", "verify", "ok" if verdict.ok else "failed: %d items" % len(verdict.failures))
if verdict.ok:
@@ -19901,9 +20309,11 @@ def note_schedule_result(db_path, row_id, status, result=""):
pass
def watch_schedule(db_path, row_id, agent_id, timeout):
def watch_schedule(db_path, row_id, agent_id, timeout, profile=""):
status, text = poll_agent(agent_id, wait=timeout + SCHEDULER_LEASE_MARGIN)
note_schedule_result(db_path, row_id, status, text)
LOG.event("INFO", "sched", "finished", "schedule #%s finished: %s" % (row_id, status), profile=profile,
meta={"id": row_id, "status": status, "result_chars": len(text or "")})
def schedule_running(last_status, updated, timeout):
@@ -19966,9 +20376,11 @@ def scheduler_tick(config, seal, runner=None):
db.commit()
skipped_cap += 1
continue
watcher = threading.Thread(target=watch_schedule, args=(config.db_path, row_id, agent_id, timeout), daemon=True)
watcher = threading.Thread(target=watch_schedule, args=(config.db_path, row_id, agent_id, timeout, profile), daemon=True)
watcher.start()
fired += 1
LOG.event("INFO", "sched", "fired", "schedule #%s fired for %s" % (row_id, profile), profile=profile,
meta={"id": row_id, "profile": profile, "every_sec": every_sec or 0})
finally:
db.close()
if skipped_cap or skipped_overlap:
@@ -20353,25 +20765,50 @@ def handle_telegram_update(agent, bot, update):
if not chat_id:
return
text = message.get("text") or ""
voice = bool(message.get("voice") or message.get("audio"))
LOG.context["chat"] = str(chat_id)
try:
outcome, reply_chars = _handle_telegram_update_inner(agent, bot, update)
except Exception as exc:
LOG.event("INFO", "telegram", "update", "telegram update failed: %s" % short_error(exc),
profile=getattr(agent, "profile", ""), bot=getattr(agent, "bot", ""), chat=str(chat_id),
meta={"chat": str(chat_id), "text_chars": len(text), "reply_chars": 0, "voice": voice,
"outcome": "error"})
raise
finally:
LOG.context.pop("chat", None)
LOG.event("INFO", "telegram", "update", "telegram update %s" % outcome,
profile=getattr(agent, "profile", ""), bot=getattr(agent, "bot", ""), chat=str(chat_id),
meta={"chat": str(chat_id), "text_chars": len(text), "reply_chars": reply_chars, "voice": voice,
"outcome": outcome})
def _handle_telegram_update_inner(agent, bot, update):
message = update.get("message") or {}
chat_id = (message.get("chat") or {}).get("id")
if not chat_id:
return "ignored", 0
text = message.get("text") or ""
if text == "/start":
telegram_send(bot, chat_id, "tai online. Send any message, /new for a fresh start.")
return
return "greeted", 0
if text == "/new":
agent.reset_history()
telegram_send(bot, chat_id, "fresh start.")
return
return "reset", 0
voice = message.get("voice") or message.get("audio")
if voice:
bot.call("sendChatAction", {"chat_id": chat_id, "action": "typing"})
text = telegram_transcribe(bot, voice.get("file_id", ""))
if text.startswith("error"):
telegram_send(bot, chat_id, text)
return
return "transcribe_error", len(text)
if not text.strip():
return
return "ignored", 0
bot.call("sendChatAction", {"chat_id": chat_id, "action": "typing"})
reply = agent.run_turn(text, capture=True)
telegram_send(bot, chat_id, reply)
return "replied", len(reply or "")
def run_telegram_bot(agent):
@@ -21302,8 +21739,12 @@ def boot(args):
print("seal upgraded from default key to TAI_PASSPHRASE")
seal.default_key = default_key and seal.enabled
store = Store(config, seal)
if LOG.enabled and LOG.db is None:
LOG.attach_store(store)
if store.schema_notes:
print("vault schema upgraded: %s" % ", ".join(store.schema_notes))
LOG.event("INFO", "vault", "schema_upgraded", "vault schema upgraded: %s" % ", ".join(store.schema_notes),
profile=config.profile, meta={"notes": ", ".join(store.schema_notes)})
applied_settings = boot_settings(config, store)
if applied_settings:
print("settings from vault: %s" % ", ".join(applied_settings))
@@ -21324,6 +21765,26 @@ def boot(args):
loaded = ""
if loaded:
print(paint(loaded, Ansi.DIM))
if getattr(args, "scheduler", False):
mode = "scheduler"
elif getattr(args, "telegram", False):
mode = "telegram"
elif getattr(args, "tui_local", False):
mode = "tui-local"
elif getattr(args, "serve_mcp", None) is not None:
mode = "serve-mcp"
elif getattr(args, "cap_speak", False):
mode = "cap-speak"
elif getattr(args, "eval_retrieval", False):
mode = "eval"
elif getattr(args, "update_models", False):
mode = "update-models"
elif getattr(args, "prompt", None):
mode = "one-shot"
else:
mode = "repl"
LOG.event("INFO", "app", "boot", "tai %s booted (%s)" % (VERSION, mode), profile=agent.profile,
meta={"version": VERSION, "home": config.home, "mode": mode})
return agent
@@ -36017,6 +36478,8 @@ def mcp_serve_handler(ctx):
self.end_headers()
return
if not ctx.authed(self.headers):
LOG.event("WARNING", "serve", "denied", "rejected unauthenticated request",
meta={"path": self.path.split("?")[0]})
self.send_response(401)
self.send_header("Content-Length", "0")
self.end_headers()
@@ -36049,6 +36512,31 @@ def mcp_serve_handler(ctx):
self._reply(answer, session)
def dispatch_message(self, message, sid):
started = time.time()
answer, session = self._dispatch_message_inner(message, sid)
method = message.get("method", "")
tool = ""
if method == "tools/call":
tool = (message.get("params") or {}).get("name") or ""
if answer is None:
outcome = "ok"
elif isinstance(answer.get("error"), dict):
outcome = "error"
elif isinstance(answer.get("result"), dict) and answer["result"].get("isError"):
outcome = "error"
else:
outcome = "ok"
try:
row = ctx.sessions.get(session or sid)
profile = row.profile if row is not None else ""
except Exception:
profile = ""
LOG.event("INFO", "serve", "request", "served %s in %dms (%s)" % (method or "?", int((time.time() - started) * 1000), outcome),
profile=profile, meta={"method": method, "tool": tool, "session": session or sid,
"exec_ms": int((time.time() - started) * 1000), "outcome": outcome})
return answer, session
def _dispatch_message_inner(self, message, sid):
method = message.get("method", "")
msg_id = message.get("id")
params = message.get("params") or {}
@@ -38345,7 +38833,7 @@ class CapAnswerer:
"""Run one isolated research turn. Returns the marker value, or
None when the turn failed, timed out, or never marked."""
with self._slots:
env = dict(os.environ, TAI_HOME=self.research_home)
env = dict(os.environ, TAI_HOME=self.research_home, TAI_CHILD="1")
try:
done = subprocess.run([sys.executable, self.tai_binary, "--auto", prompt],
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
@@ -38999,7 +39487,19 @@ class CapFleet:
if self.budget_usd is not None and self.spent_usd() >= self.budget_usd:
raise CapError("cap_budget_exceeded", "fleet spent %.4f of %.4f USD" % (self.spent_usd(), self.budget_usd))
self._log("turn.start", recipient=urn, text=text)
started = time.time()
result = self.sessions[urn].ask(text, timeout)
state = getattr(result, "state", "?")
if state == "done":
outcome = "ok"
elif state == "timeout":
outcome = "timeout"
else:
outcome = "error"
LOG.event("INFO", "cap", "turn", "cap turn to %s finished: %s" % (urn, state),
meta={"target": urn, "payload_chars": len(text or ""),
"exec_ms": int((time.time() - started) * 1000),
"reply_chars": len(getattr(result, "text", "") or ""), "outcome": outcome})
self._log("turn.end", recipient=urn, text="%s, %d events" % (result.state, len(result.events)))
self.pump()
return result
@@ -39282,6 +39782,8 @@ def cap_run_main(argv):
print("cap run: %s" % exc, file=sys.stderr)
return 2
fleet = CapFleet(budget_usd=plan["budget_usd"])
LOG.event("INFO", "cap", "run_start", "cap run started (%d sessions)" % len(plan["sessions"]),
meta={"sessions": len(plan["sessions"])})
auto = args.auto_answer or plan["auto_answer"]
max_risk = plan["max_risk"] if args.max_risk == "medium" else args.max_risk
fleet.set_answerer(cap_build_answerer(auto, args.allow_tool + plan["allow_tools"],
@@ -39426,6 +39928,8 @@ def build_parser():
parser.add_argument("--yolo", action="store_true")
parser.add_argument("--auto", action="store_true")
parser.add_argument("--continue", dest="resume", action="store_true")
parser.add_argument("--debug", nargs="?", const="file", default=None, choices=("file", "stdout", "stderr"), metavar="SINK", help="extensive debug logging to <home>/debug.log (file), stdout, or stderr, plus the vault logs table (see logging.md)")
parser.add_argument("--debug-format", default=None, choices=("sentence", "json"), metavar="FORMAT", help="debug log line format, sentence or json (default sentence, or TAI_DEBUG_FORMAT)")
parser.add_argument("--env", default=None)
parser.add_argument("--no-opencode", action="store_true", help="skip the opencode CLI backends for this run (same as TAI_OPENCODE=0)")
parser.add_argument("--update-models", action="store_true", help="rediscover the free model lists of every provider, drop models that are gone, and exit")
@@ -39559,6 +40063,13 @@ def update_models_command(args):
def main(argv=None):
args = build_parser().parse_args(argv)
debug_sink = args.debug or (os.environ.get("TAI_DEBUG") or "").strip().lower()
if debug_sink in ("file", "stdout", "stderr"):
debug_format = args.debug_format or (os.environ.get("TAI_DEBUG_FORMAT") or "").strip().lower()
if debug_format not in ("sentence", "json"):
debug_format = "sentence"
home = os.environ.get("TAI_HOME") or os.path.join(os.path.expanduser("~"), ".tai")
LOG.configure(debug_sink, debug_format, home)
if args.install:
return install_self()
if args.install_telegram:
test_tai.py
+206
View File
@@ -16939,5 +16939,211 @@ class McpDocTests(unittest.TestCase):
self.assertNotIn(chr(0x2014), self.doc)
class LoggerTests(unittest.TestCase):
def setUp(self):
tai.LOG.reset()
self.tmp = tempfile.TemporaryDirectory()
self.home = os.path.join(self.tmp.name, "home")
os.makedirs(self.home)
self.env = mock.patch.dict(os.environ, {"HOME": self.home, "TAI_HOME": os.path.join(self.home, ".tai")})
self.env.start()
os.environ.pop("TAI_CHILD", None)
class FakeArgs:
profile = "t"
yes = True
self.config = tai.Config(FakeArgs())
self.seal = tai.Seal(self.config.home, "log-test-1")
self.store = tai.Store(self.config, self.seal)
def tearDown(self):
tai.LOG.reset()
self.store.close()
self.env.stop()
self.tmp.cleanup()
os.environ.pop("TAI_HOME", None)
def logfile(self):
with open(os.path.join(self.config.home, "debug.log"), encoding="utf-8") as handle:
return handle.read()
def test_sentence_format_and_db(self):
tai.LOG.configure("file", "sentence", self.config.home, self.store)
tai.LOG.event("INFO", "tool", "call", "tool x finished", profile="t", bot="main",
meta={"name": "x", "exec_ms": 5, "note": "two words"})
line = self.logfile().strip()
self.assertRegex(line, r"^\S+ INFO run=\S+ profile=t bot=main chat=\"\" tool\.call ")
self.assertIn("exec_ms=5", line)
self.assertIn('note="two words"', line)
self.assertTrue(line.endswith(":: tool x finished"))
tai.LOG.db.commit()
rows = self.store.log_query()
self.assertEqual(len(rows), 1)
self.assertEqual(rows[0]["component"], "tool")
self.assertEqual(rows[0]["meta"]["exec_ms"], 5)
def test_json_format(self):
tai.LOG.configure("file", "json", self.config.home, self.store)
tai.LOG.event("WARNING", "llm", "failed", "p/m failed", profile="t", meta={"error": "boom"})
line = self.logfile().strip()
data = json.loads(line)
self.assertEqual(data["level"], "WARNING")
self.assertEqual(data["component"], "llm")
self.assertEqual(data["meta"], {"error": "boom"})
self.assertEqual(list(data), sorted(data))
def test_wipe_and_append(self):
path = os.path.join(self.config.home, "debug.log")
os.makedirs(self.config.home, exist_ok=True)
with open(path, "w", encoding="utf-8") as handle:
handle.write("old\n")
tai.LOG.configure("file", "sentence", self.config.home, self.store)
tai.LOG.event("INFO", "app", "boot", "fresh", profile="t")
self.assertNotIn("old", self.logfile())
tai.LOG.reset()
with open(path, "w", encoding="utf-8") as handle:
handle.write("old\n")
with mock.patch.dict(os.environ, {"TAI_CHILD": "1"}):
tai.LOG.configure("file", "sentence", self.config.home, self.store)
tai.LOG.event("INFO", "app", "boot", "fresh", profile="t")
self.assertIn("old", self.logfile())
self.assertIn("fresh", self.logfile())
def test_redact_and_mask(self):
self.store.save_secret("wifi", "s3cr3t-value", profile="t")
tai.LOG.configure("file", "sentence", self.config.home, self.store)
tai.LOG.event("INFO", "agent", "turn_start", "do s3cr3t-value now", profile="t", meta={"goal": "s3cr3t-value"})
blob = self.logfile()
self.assertNotIn("s3cr3t-value", blob)
self.assertIn("[redacted:wifi]", blob)
agent = mock.Mock()
agent.store = self.store
agent.profile = "t"
agent.bot = "main"
tools = tai.Tools(agent)
tools.log_call("store_secret", {"name": "wifi", "value": "brand-new-secret"}, time.time(), "ok", "stored")
blob = self.logfile()
self.assertNotIn("brand-new-secret", blob)
self.assertIn("***", blob)
def test_prune_and_fts(self):
tai.LOG.configure("file", "sentence", self.config.home, self.store)
for pos in range(5):
tai.LOG.event("INFO", "tool", "call", "call %d" % pos, profile="t")
tai.LOG.db.commit()
with mock.patch.object(tai, "LOG_DB_KEEP", 3):
tai.LOG.prune()
self.assertEqual(len(self.store.log_query(limit=100)), 3)
fts = self.store.db.execute("SELECT COUNT(*) FROM fts_docs WHERE kind = 'log'").fetchone()[0]
self.assertEqual(fts, 3)
def test_never_raise(self):
tai.LOG.configure("file", "sentence", self.config.home, self.store)
with mock.patch.object(tai.DebugLog, "_open_handle", side_effect=OSError("disk gone")):
tai.LOG.event("INFO", "app", "boot", "x", profile="t")
self.assertEqual(tai.LOG.dropped, 1)
def test_disabled_noop(self):
tai.LOG.event("INFO", "app", "boot", "x", profile="t")
self.assertFalse(os.path.exists(os.path.join(self.config.home, "debug.log")))
def test_dispatch_hook(self):
tai.LOG.configure("file", "sentence", self.config.home, self.store)
agent = mock.Mock()
agent.store = self.store
agent.profile = "t"
agent.bot = "main"
tools = tai.Tools(agent)
out = tools.dispatch("nope_tool_xyz", {})
self.assertIn("unknown tool", out)
tai.LOG.db.commit()
rows = self.store.log_query(component="tool", event="call")
self.assertEqual(len(rows), 1)
self.assertEqual(rows[0]["meta"]["outcome"], "unknown_tool")
def test_completion_hook(self):
tai.LOG.configure("file", "sentence", self.config.home, self.store)
chat = tai.ChatClient(self.config)
cand = {"provider": "p", "model": "m"}
messages = [{"role": "system", "content": "sys"}, {"role": "user", "content": "hi"}]
reply = {"role": "assistant", "content": "yo", "reasoning": "", "tool_calls": [], "backend": "p"}
with mock.patch.object(tai.ChatClient, "call_candidate", return_value=reply):
self.assertEqual(chat.try_candidate(cand, messages, None, {"model": "x"}, None, False, []), reply)
tai.LOG.db.commit()
rows = self.store.log_query(component="llm", event="completion")
self.assertEqual(len(rows), 1)
meta = rows[0]["meta"]
self.assertEqual(meta["model"], "m")
self.assertEqual(meta["assistant_text"], "yo")
self.assertEqual(meta["tokens_total"], meta["tokens_in"] + meta["tokens_out"])
self.assertIsNone(meta["cost_usd"])
with mock.patch.object(tai.ChatClient, "call_candidate", side_effect=tai.BackendError("boom")):
self.assertIsNone(chat.try_candidate(cand, messages, None, {"model": "x"}, None, False, []))
tai.LOG.db.commit()
rows = self.store.log_query(component="llm", event="failed")
self.assertEqual(len(rows), 1)
self.assertEqual(rows[0]["level"], "WARNING")
def test_approval_hook(self):
tai.LOG.configure("file", "sentence", self.config.home, self.store)
agent = tai.Agent(self.config, self.store, persist=False, quiet=True, yolo=True)
self.assertTrue(agent.ask_approval("rm -rf /"))
tai.LOG.db.commit()
rows = self.store.log_query(component="approval", event="decision")
self.assertEqual(len(rows), 1)
self.assertTrue(rows[0]["meta"]["decision"])
self.assertEqual(rows[0]["meta"]["mode"], "auto")
def test_turn_wrapper(self):
tai.LOG.configure("file", "sentence", self.config.home, self.store)
agent = tai.Agent(self.config, self.store, persist=False, quiet=True)
reply = {"role": "assistant", "content": "done", "reasoning": "", "tool_calls": [], "backend": "x"}
with mock.patch.object(agent.chat, "complete", return_value=reply):
self.assertEqual(agent.run_turn("hi", capture=True), "done")
tai.LOG.db.commit()
starts = self.store.log_query(component="agent", event="turn_start")
ends = self.store.log_query(component="agent", event="turn_end")
self.assertEqual(len(starts), 1)
self.assertEqual(len(ends), 1)
self.assertEqual(ends[0]["meta"]["outcome"], "ok")
self.assertEqual(ends[0]["meta"]["reply_chars"], 4)
def test_excepthook(self):
tai.LOG.configure("file", "sentence", self.config.home, self.store)
tai.LOG.old_excepthook = None
tai.LOG._excepthook(ValueError, ValueError("boom"), None)
tai.LOG._excepthook(KeyboardInterrupt, KeyboardInterrupt(), None)
tai.LOG.db.commit()
rows = self.store.log_query(component="app", event="exception")
self.assertEqual(len(rows), 1)
self.assertIn("ValueError", rows[0]["meta"]["error"])
class LoggingDocTests(unittest.TestCase):
def setUp(self):
with open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "logging.md"), encoding="utf-8") as handle:
self.doc = handle.read()
def test_doc_names_everything(self):
for const in ("LOG_DB_KEEP", "LOG_DB_PRUNE_EVERY", "LOG_MSG_MAX",
"LOG_TEXT_MAX", "LOG_PREVIEW", "LOG_META_MAX", "LOG_MASKED_ARGS"):
self.assertIn(const, self.doc, const)
self.assertTrue(hasattr(tai, const), const)
for event in ("app.boot", "app.exception", "agent.turn_start", "agent.turn_end", "agent.verified",
"agent.compact", "llm.completion", "llm.failed", "llm.round_retry", "llm.fatal",
"tool.call", "approval.decision", "vault.record_saved", "vault.record_deleted",
"vault.secret_saved", "vault.secret_deleted", "vault.schedule_saved", "vault.schedule_removed",
"vault.seal_rotated", "vault.schema_upgraded", "sched.fired", "sched.finished",
"serve.request", "serve.denied", "cap.run_start", "cap.turn", "telegram.update"):
self.assertIn(event, self.doc, event)
for name in ("logs(id INTEGER PRIMARY KEY", "idx_logs_profile", "idx_logs_bot", "idx_logs_chat",
"idx_logs_level", "idx_logs_event", "idx_logs_run", "trg_logs_ai", "trg_logs_ad",
"--debug", "--debug-format", "sentence", "json", "TAI_CHILD", "TAI_DEBUG",
"DebugLog", "log_query"):
self.assertIn(name, self.doc, name)
self.assertNotIn(chr(0x2014), self.doc)
if __name__ == "__main__":
unittest.main()