| 1 | import { PET_MAX_SECONDS } from './pet-sim.js'; |
| 2 | import type { Category, WhaleEvent } from './model.js'; |
| 3 | import { toolActivity, toolCategory } from './codewhale.js'; |
| 4 | import { compilePetTelemetry, PET_BIN_MS, type PetBucket } from './pet-telemetry.js'; |
| 5 | |
| 6 | /** Read-only adapter for codewhale_protocol::EventMsg metadata. The foreground |
| 7 | * Engine is the event owner. This replaces no turn loop: it only translates |
| 8 | * lifecycle observations to event-v1 for the same pet bucketer used by imports. |
| 9 | * Text, inputs and results are neither accepted nor retained. */ |
| 10 | export class PetEngineTelemetry { |
| 11 | private events: WhaleEvent[] = []; |
| 12 | private active = new Map<string, WhaleEvent>(); |
| 13 | private waiting: WhaleEvent | undefined; |
| 14 | private sequence = 0; |
| 15 | private lastTime = 0; |
| 16 | |
| 17 | /** Ephemeral receipts. Replay tapes retain measured categories, not tool |
| 18 | * names. Restoring/disconnecting clears these captions. Never mutates world. */ |
| 19 | activity(at: number) { |
| 20 | const fresh = (e: WhaleEvent) => at >= e.startTime && at - e.endTime <= PET_BIN_MS * 2; |
| 21 | const spans = [...this.active].filter(([, e]) => fresh(e)); |
| 22 | const parallel = spans.filter(([key]) => key.startsWith('agent:')).length; |
| 23 | const cue = ([key, e]: [string, WhaleEvent]) => ({ |
| 24 | ...(key.startsWith('tool:') ? toolActivity(e.name) : key.startsWith('thinking:') |
| 25 | ? { kind: 'thinking', label: 'Thinking' } : key.startsWith('agent:') |
| 26 | ? { kind: 'delegating', label: 'Coordinating agents' } : { kind: 'responding', label: 'Writing the response' }), |
| 27 | tool: key.startsWith('tool:') ? e.name.replace(/[^a-zA-Z0-9_.:-]/g, '').slice(0, 96) : null, |
| 28 | sinceMs: e.startTime, |
| 29 | }); |
| 30 | const active = spans.filter(([key]) => !key.startsWith('agent:')).slice(-4).reverse().map(cue); |
| 31 | const error = [...this.events].reverse().find(e => e.category === 'error' && fresh(e)); |
| 32 | const primary = this.waiting && fresh(this.waiting) |
| 33 | ? { kind: 'waiting', label: 'Waiting for you', tool: null, sinceMs: this.waiting.startTime } |
| 34 | : error ? { kind: 'error', label: 'An operation failed', tool: null, sinceMs: error.startTime } |
| 35 | : active[0] ?? (parallel ? cue(spans.find(([key]) => key.startsWith('agent:'))!) |
| 36 | : { kind: 'unknown', label: 'Activity unobserved', tool: null, sinceMs: at }); |
| 37 | return { ...primary, observed: primary.kind !== 'unknown', parallel, active }; |
| 38 | } |
| 39 | |
| 40 | private add(name: string, category: Category, at: number, agentId = 'parent', continuation = false): WhaleEvent { |
| 41 | if (this.events.length >= 8192) throw new Error('Pet Engine observation window is full.'); |
| 42 | const e: WhaleEvent = { schemaVersion: 1, id: `engine:${this.sequence++}`, traceId: 'foreground', |
| 43 | startTime: at, endTime: at, name, category, agentId, status: 'running', |
| 44 | attributes: continuation ? { 'whalesong.continuation': true } : {} }; |
| 45 | this.events.push(e); return e; |
| 46 | } |
| 47 | |
| 48 | private pulse(key: string, at: number): void { |
| 49 | const e = this.active.get(key); |
| 50 | if (!e) return; |
| 51 | // A resumed stream does not assert coverage across its silent interval. |
| 52 | if (at - e.endTime > PET_BIN_MS * 2) { |
| 53 | this.active.set(key, this.add(e.name, e.category, at, e.agentId, true)); |
| 54 | } else e.endTime = at; |
| 55 | } |
| 56 | |
| 57 | /** Transactional batch copy; failed validation cannot accept half a packet. */ |
| 58 | clone(): PetEngineTelemetry { |
| 59 | const next = new PetEngineTelemetry(); |
| 60 | const copy = <T>(value: T): T => JSON.parse(JSON.stringify(value)) as T; |
| 61 | next.events = copy(this.events); |
| 62 | const spans = new Map(next.events.map(event => [event.id, event])); |
| 63 | // Active and waiting spans must still reference their journal entry so a |
| 64 | // later heartbeat extends the coverage consumed by bucket(). |
| 65 | const span = (event: WhaleEvent) => spans.get(event.id) ?? copy(event); |
| 66 | next.active = new Map(Array.from(this.active, ([key, event]) => [key, span(event)])); |
| 67 | next.waiting = this.waiting ? span(this.waiting) : undefined; |
| 68 | next.sequence = this.sequence; next.lastTime = this.lastTime; |
| 69 | return next; |
| 70 | } |
| 71 | |
| 72 | observe(value: unknown, at: number): void { |
| 73 | if (!Number.isFinite(at) || at < this.lastTime || at > PET_MAX_SECONDS * 1000) throw new Error('Invalid Engine pet clock.'); |
| 74 | if (!value || typeof value !== 'object' || Array.isArray(value)) throw new Error('Invalid Engine pet metadata.'); |
| 75 | const e = value as Record<string, unknown>; |
| 76 | const allowed = ['event', 'index', 'channel', 'tool_call_id', 'tool_name', 'id', 'worker_status', 'failed']; |
| 77 | if (Object.keys(e).some(k => !allowed.includes(k)) || typeof e.event !== 'string' |
| 78 | || Object.values(e).some(v => typeof v === 'string' && v.length > 4096) |
| 79 | || e.channel !== undefined && !['text', 'reasoning'].includes(e.channel as string) |
| 80 | || ['tool_call_id', 'tool_name', 'id', 'worker_status'].some(k => e[k] !== undefined && typeof e[k] !== 'string') |
| 81 | || e.failed !== undefined && typeof e.failed !== 'boolean' |
| 82 | || e.index !== undefined && (!Number.isSafeInteger(e.index) || (e.index as number) < 0)) |
| 83 | throw new Error('Invalid Engine pet metadata fields.'); |
| 84 | this.lastTime = at; |
| 85 | this.events = this.events.filter(span => span.endTime >= at - 12_800); |
| 86 | const id = (field: string) => { const s = e[field]; if (typeof s !== 'string' || !s) throw new Error(`Missing Engine ${field}.`); return s; }; |
| 87 | const index = () => { if (!Number.isSafeInteger(e.index)) throw new Error('Missing Engine index.'); return String(e.index); }; |
| 88 | const start = (key: string, name: string, category: Category, agentId?: string) => { |
| 89 | if (this.active.size >= 256 && !this.active.has(key)) throw new Error('Too many active Engine pet spans.'); |
| 90 | this.active.set(key, this.add(name, category, at, agentId)); |
| 91 | }; |
| 92 | const finish = (key: string) => { this.pulse(key, at); this.active.delete(key); }; |
| 93 | switch (e.event) { |
| 94 | case 'turn_started': this.active.clear(); this.waiting = undefined; break; |
| 95 | case 'message_started': start(`message:${index()}`, 'assistant_message', 'communication'); this.waiting = undefined; break; |
| 96 | case 'thinking_started': start(`thinking:${index()}`, 'thinking', 'reasoning'); this.waiting = undefined; break; |
| 97 | case 'response_delta': { |
| 98 | const reasoning = e.channel === 'reasoning'; |
| 99 | const key = `${reasoning ? 'thinking' : 'message'}:${index()}`; |
| 100 | if (!this.active.has(key)) start(key, reasoning ? 'thinking' : 'assistant_message', reasoning ? 'reasoning' : 'communication'); |
| 101 | else this.pulse(key, at); |
| 102 | this.waiting = undefined; break; |
| 103 | } |
| 104 | case 'message_complete': finish(`message:${index()}`); break; |
| 105 | case 'thinking_complete': finish(`thinking:${index()}`); break; |
| 106 | case 'tool_call_started': start(`tool:${id('tool_call_id')}`, id('tool_name'), toolCategory(id('tool_name'))); this.waiting = undefined; break; |
| 107 | case 'tool_call_heartbeat': for (const key of this.active.keys()) if (key.startsWith('tool:')) this.pulse(key, at); break; |
| 108 | case 'tool_call_complete': finish(`tool:${id('tool_call_id')}`); this.waiting = undefined; break; |
| 109 | case 'agent_spawned': start(`agent:${id('id')}`, 'agent', 'agent', id('id')); break; |
| 110 | case 'agent_progress': { |
| 111 | const key = `agent:${id('id')}`; |
| 112 | if (['completed', 'failed', 'cancelled', 'interrupted', 'budget_exhausted'].includes(e.worker_status as string)) { finish(key); break; } |
| 113 | if (!this.active.has(key)) start(key, 'agent', 'agent', id('id')); else this.pulse(key, at); |
| 114 | break; |
| 115 | } |
| 116 | case 'agent_complete': finish(`agent:${id('id')}`); break; |
| 117 | case 'approval_required': case 'user_input_required': this.waiting = this.add('human', 'human', at); this.waiting.status = 'pending'; break; |
| 118 | case 'turn_complete': this.active.clear(); this.waiting = undefined; break; |
| 119 | case 'error': break; |
| 120 | default: throw new Error('Unsupported Engine pet event.'); |
| 121 | } |
| 122 | // Receipt time is the error onset; never rewrite the operation's old start. |
| 123 | if (e.event === 'error' || e.failed === true) this.add('error', 'error', at).status = 'error'; |
| 124 | } |
| 125 | |
| 126 | /** Waiting coverage comes from the existing typed shell's current request. |
| 127 | * It can extend a witnessed request, never invent one on a mid-turn attach. */ |
| 128 | confirmWaiting(at: number, waiting: boolean): void { |
| 129 | if (!waiting) { this.waiting = undefined; return; } |
| 130 | if (this.waiting && at >= this.waiting.endTime) this.waiting.endTime = at; |
| 131 | } |
| 132 | |
| 133 | bucket(sequence: number): PetBucket { |
| 134 | const end = (sequence + 1) * PET_BIN_MS; |
| 135 | const input = this.events.filter(e => e.startTime < end && e.endTime >= end - 12_400) |
| 136 | .map(e => ({ ...e, endTime: Math.min(e.endTime, end) })); |
| 137 | return compilePetTelemetry(input, end, sequence)[0]; |
| 138 | } |
| 139 | } |
| 140 |