INNER CODE UNIT · Python
verify
omnimind-ai/OmniBot · scripts/assert-agent-init.py:12
def verify(db, after, expected="end_turn", attachment=None):
rows = [(i, c, t, json.loads(p)) for i, c, t, p in db.execute(
'SELECT id,conversationId,entryType,payloadJson FROM agent_conversation_entries WHERE id>? ORDER BY id', (after,))]
users = [r for r in rows if r[2] == 'user_message']
if not users: raise InitPending('Waiting for init admission')
assert len(users) == 1, 'Init admission missing or duplicated; another send invalidates this observation'
user = users[0]
text = user[3].get('content', {}).get('text', '')
assert text == '/init' or text.startswith('Please analyze this repository and create or update an AGENTS.md file'), 'Not an init submission'
if attachment:
refs = user[3].get('content', {}).get('attachments', [])
assert len(refs) == 1 and refs[0].get('name') == attachment and refs[0].get('path'), 'Init attachment missing or duplicated'
items = [p for i, c, t, p in rows if c == user[1] and i > user[0]]
if not items: raise InitPending('No init output yet')
identities = {(p.get('streamMeta', {}).get('sessionId'), p.get('streamMeta', {}).get('turnId')) for p in items}
assert len(identities) == 1 and all(next(iter(identities))), 'Missing or mixed canonical turn identity'
if expected != 'active' and any(p.get('isLoading', False) for p in items): raise InitPending('Init is still loading')
stops = {p.get('streamMeta', {}).get('stopReason') for p in items} - {None, ''}