Merge master 36fe4785 into feat/issue-660-session-checkpoint-schema
Base-sync of the #660 durable session checkpoint schema branch onto current master. The only conflicting file was control_plane_db.py, where master added the #651 usage_events table, migration, and indexes while this branch added the Resolution keeps both sides in full. Verified against both merge parents: the resolved file contains every line of master's control_plane_db.py with zero removals, plus exactly the #660 additions (session_checkpoints table, its two indexes, _session_checkpoint_row, write_session_checkpoint, get_session_checkpoint, list_session_checkpoints, reconcile_session_checkpoint, and checkpoint_completeness). Conflicts: control_plane_db.py Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
Claude Opus 4.8
@@ -238,6 +238,34 @@ CREATE INDEX IF NOT EXISTS idx_session_checkpoints_session
|
||||
ON session_checkpoints(remote, org, repo, session_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_session_checkpoints_work
|
||||
ON session_checkpoints(remote, org, repo, work_kind, work_number);
|
||||
|
||||
-- Model usage, token cost, latency, and performance events (#651)
|
||||
CREATE TABLE IF NOT EXISTS usage_events (
|
||||
usage_id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
session_id TEXT,
|
||||
remote TEXT NOT NULL DEFAULT 'dadeschools',
|
||||
org TEXT NOT NULL DEFAULT '',
|
||||
repo TEXT NOT NULL DEFAULT '',
|
||||
project_id TEXT,
|
||||
role TEXT NOT NULL DEFAULT 'unknown',
|
||||
model TEXT NOT NULL DEFAULT 'unknown',
|
||||
issue_number INTEGER,
|
||||
pr_number INTEGER,
|
||||
stage TEXT NOT NULL DEFAULT 'unknown',
|
||||
input_tokens INTEGER,
|
||||
output_tokens INTEGER,
|
||||
total_tokens INTEGER,
|
||||
estimated_cost_usd REAL,
|
||||
latency_ms INTEGER,
|
||||
duration_ms INTEGER,
|
||||
status TEXT NOT NULL DEFAULT 'success',
|
||||
metadata TEXT,
|
||||
created_at TEXT NOT NULL
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_usage_events_scope ON usage_events(remote, org, repo);
|
||||
CREATE INDEX IF NOT EXISTS idx_usage_events_role_model ON usage_events(role, model);
|
||||
CREATE INDEX IF NOT EXISTS idx_usage_events_stage ON usage_events(stage);
|
||||
"""
|
||||
|
||||
|
||||
@@ -391,6 +419,7 @@ class ControlPlaneDB:
|
||||
self._migrate_incident_links_null_scope(conn)
|
||||
self._migrate_lease_lifecycle_columns(conn)
|
||||
self._migrate_session_ownership_columns(conn)
|
||||
self._migrate_usage_events_table(conn)
|
||||
conn.execute(
|
||||
"INSERT OR REPLACE INTO schema_meta(key, value) VALUES (?, ?)",
|
||||
("schema_version", str(SCHEMA_VERSION)),
|
||||
@@ -570,6 +599,207 @@ class ControlPlaneDB:
|
||||
f"UPDATE incident_links SET {col} = '' WHERE {col} IS NULL"
|
||||
)
|
||||
|
||||
def _migrate_usage_events_table(self, conn: sqlite3.Connection) -> None:
|
||||
"""Create usage_events table and indexes if they do not exist (#651)."""
|
||||
conn.execute("""
|
||||
CREATE TABLE IF NOT EXISTS usage_events (
|
||||
usage_id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
session_id TEXT,
|
||||
remote TEXT NOT NULL DEFAULT 'dadeschools',
|
||||
org TEXT NOT NULL DEFAULT '',
|
||||
repo TEXT NOT NULL DEFAULT '',
|
||||
project_id TEXT,
|
||||
role TEXT NOT NULL DEFAULT 'unknown',
|
||||
model TEXT NOT NULL DEFAULT 'unknown',
|
||||
issue_number INTEGER,
|
||||
pr_number INTEGER,
|
||||
stage TEXT NOT NULL DEFAULT 'unknown',
|
||||
input_tokens INTEGER,
|
||||
output_tokens INTEGER,
|
||||
total_tokens INTEGER,
|
||||
estimated_cost_usd REAL,
|
||||
latency_ms INTEGER,
|
||||
duration_ms INTEGER,
|
||||
status TEXT NOT NULL DEFAULT 'success',
|
||||
metadata TEXT,
|
||||
created_at TEXT NOT NULL
|
||||
);
|
||||
""")
|
||||
conn.execute("CREATE INDEX IF NOT EXISTS idx_usage_events_scope ON usage_events(remote, org, repo);")
|
||||
conn.execute("CREATE INDEX IF NOT EXISTS idx_usage_events_role_model ON usage_events(role, model);")
|
||||
conn.execute("CREATE INDEX IF NOT EXISTS idx_usage_events_stage ON usage_events(stage);")
|
||||
|
||||
# #651 retention: cap growth so unauthenticated or high-volume ingest
|
||||
# cannot DoS the control-plane DB (PR #876 F3). Applied after every write.
|
||||
USAGE_EVENTS_MAX_ROWS = 10_000
|
||||
USAGE_EVENTS_RETENTION_DAYS = 90
|
||||
|
||||
def record_usage_event(
|
||||
self,
|
||||
*,
|
||||
session_id: str | None = None,
|
||||
remote: str = "dadeschools",
|
||||
org: str = "",
|
||||
repo: str = "",
|
||||
project_id: str | None = None,
|
||||
role: str = "unknown",
|
||||
model: str = "unknown",
|
||||
issue_number: int | None = None,
|
||||
pr_number: int | None = None,
|
||||
stage: str = "unknown",
|
||||
input_tokens: int | None = None,
|
||||
output_tokens: int | None = None,
|
||||
total_tokens: int | None = None,
|
||||
estimated_cost_usd: float | None = None,
|
||||
latency_ms: int | None = None,
|
||||
duration_ms: int | None = None,
|
||||
status: str = "success",
|
||||
metadata: str | dict[str, Any] | None = None,
|
||||
created_at: str | None = None,
|
||||
) -> int:
|
||||
"""Record a model usage, token cost, latency, or stage performance event (#651)."""
|
||||
ts = created_at or _ts()
|
||||
meta_str: str | None = None
|
||||
if metadata is not None:
|
||||
from webui import console_redaction
|
||||
redacted_meta = console_redaction.redact_payload(metadata)
|
||||
if isinstance(redacted_meta, str):
|
||||
meta_str = redacted_meta
|
||||
else:
|
||||
try:
|
||||
meta_str = json.dumps(redacted_meta, default=str)
|
||||
except Exception:
|
||||
meta_str = str(redacted_meta)
|
||||
|
||||
if total_tokens is None and (input_tokens is not None or output_tokens is not None):
|
||||
total_tokens = (input_tokens or 0) + (output_tokens or 0)
|
||||
|
||||
with self._tx(immediate=True) as conn:
|
||||
cursor = conn.execute(
|
||||
"""
|
||||
INSERT INTO usage_events (
|
||||
session_id, remote, org, repo, project_id, role, model,
|
||||
issue_number, pr_number, stage, input_tokens, output_tokens,
|
||||
total_tokens, estimated_cost_usd, latency_ms, duration_ms,
|
||||
status, metadata, created_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
(
|
||||
session_id,
|
||||
remote,
|
||||
org,
|
||||
repo,
|
||||
project_id,
|
||||
role,
|
||||
model,
|
||||
issue_number,
|
||||
pr_number,
|
||||
stage,
|
||||
input_tokens,
|
||||
output_tokens,
|
||||
total_tokens,
|
||||
estimated_cost_usd,
|
||||
latency_ms,
|
||||
duration_ms,
|
||||
status,
|
||||
meta_str,
|
||||
ts,
|
||||
),
|
||||
)
|
||||
usage_id = cursor.lastrowid
|
||||
self._enforce_usage_events_retention(conn)
|
||||
return usage_id
|
||||
|
||||
def _enforce_usage_events_retention(self, conn: sqlite3.Connection) -> None:
|
||||
"""Drop aged and excess usage_events rows (PR #876 F3)."""
|
||||
# Age-based: ISO-8601 UTC timestamps compare lexicographically.
|
||||
cutoff = (
|
||||
datetime.now(timezone.utc)
|
||||
- timedelta(days=int(self.USAGE_EVENTS_RETENTION_DAYS))
|
||||
).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
conn.execute(
|
||||
"DELETE FROM usage_events WHERE created_at < ?",
|
||||
(cutoff,),
|
||||
)
|
||||
# Count-based: keep the newest USAGE_EVENTS_MAX_ROWS by usage_id.
|
||||
max_rows = int(self.USAGE_EVENTS_MAX_ROWS)
|
||||
if max_rows > 0:
|
||||
conn.execute(
|
||||
"""
|
||||
DELETE FROM usage_events
|
||||
WHERE usage_id NOT IN (
|
||||
SELECT usage_id FROM usage_events
|
||||
ORDER BY usage_id DESC
|
||||
LIMIT ?
|
||||
)
|
||||
""",
|
||||
(max_rows,),
|
||||
)
|
||||
|
||||
def query_usage_events(
|
||||
self,
|
||||
*,
|
||||
remote: str | None = None,
|
||||
org: str | None = None,
|
||||
repo: str | None = None,
|
||||
project_id: str | None = None,
|
||||
role: str | None = None,
|
||||
model: str | None = None,
|
||||
issue_number: int | None = None,
|
||||
pr_number: int | None = None,
|
||||
stage: str | None = None,
|
||||
session_id: str | None = None,
|
||||
limit: int = 500,
|
||||
offset: int = 0,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Query stored usage events matching filters (#651)."""
|
||||
conditions = []
|
||||
params = []
|
||||
if remote:
|
||||
conditions.append("remote = ?")
|
||||
params.append(remote)
|
||||
if org:
|
||||
conditions.append("org = ?")
|
||||
params.append(org)
|
||||
if repo:
|
||||
conditions.append("repo = ?")
|
||||
params.append(repo)
|
||||
if project_id:
|
||||
conditions.append("project_id = ?")
|
||||
params.append(project_id)
|
||||
if role:
|
||||
conditions.append("role = ?")
|
||||
params.append(role)
|
||||
if model:
|
||||
conditions.append("model = ?")
|
||||
params.append(model)
|
||||
if issue_number is not None:
|
||||
conditions.append("issue_number = ?")
|
||||
params.append(issue_number)
|
||||
if pr_number is not None:
|
||||
conditions.append("pr_number = ?")
|
||||
params.append(pr_number)
|
||||
if stage:
|
||||
conditions.append("stage = ?")
|
||||
params.append(stage)
|
||||
if session_id:
|
||||
conditions.append("session_id = ?")
|
||||
params.append(session_id)
|
||||
|
||||
where_clause = f"WHERE {' AND '.join(conditions)}" if conditions else ""
|
||||
sql = f"""
|
||||
SELECT * FROM usage_events
|
||||
{where_clause}
|
||||
ORDER BY usage_id ASC
|
||||
LIMIT ? OFFSET ?
|
||||
"""
|
||||
params.extend([limit, offset])
|
||||
|
||||
with self._tx(immediate=False) as conn:
|
||||
cursor = conn.execute(sql, params)
|
||||
rows = cursor.fetchall()
|
||||
return [dict(row) for row in rows]
|
||||
|
||||
# ── sessions ──────────────────────────────────────────────────────────
|
||||
|
||||
def upsert_session(
|
||||
|
||||
Reference in New Issue
Block a user