INNER CODE UNIT · Python
verify_live_checkpoint
omnimind-ai/OmniBot · scripts/assert-agent-context-checkpoint.py:6
def verify_live_checkpoint(db, marker):
assert re.fullmatch(r'OOB_LIVE_AUTO_COMPACT_\d+', marker), 'Automatic compaction test marker required'
users = db.execute("SELECT id,conversationId FROM agent_conversation_entries WHERE entryType='user_message' AND instr(payloadJson,?)>0", (marker,)).fetchall()
assert len(users) == 1, 'Missing or duplicate user admission'
user, conversation = users[0]
summary, cutoff, revision = db.execute('SELECT contextSummary,contextSummaryCutoffEntryDbId,contextSummaryUpdatedAt FROM conversations WHERE id=?', (conversation,)).fetchone()
assert summary and cutoff and revision, 'No durable summary'
assert cutoff > user, 'Old checkpoint does not prove same-task automatic compaction'
next_user = db.execute("SELECT min(id) FROM agent_conversation_entries WHERE conversationId=? AND entryType='user_message' AND id>?", (conversation,user)).fetchone()[0]
assert next_user is None or cutoff < next_user, 'Later task checkpoint does not prove this task compacted'
row = db.execute('SELECT entryType,payloadJson FROM agent_conversation_entries WHERE id=? AND conversationId=?', (cutoff, conversation)).fetchone()
assert row and row[0] == 'tool_event', 'Checkpoint does not identify a completed tool boundary'
payload = json.loads(row[1])
assert payload.get('toolCallId') and payload.get('sessionId') and payload.get('success') is True, 'Checkpoint tool identity or completion missing'
return {'conversationId':conversation,'summarySha256':hashlib.sha256(summary.encode()).hexdigest(),'cutoff':cutoff,'revision':revision}
if __name__ == '__main__':
serial, marker = sys.argv[1:]