import { RoomEvent, type Room } from '@livekit/rtc-node'; import type { EngineerContext, Collision, Intervention, DataMessage } from '@podman/shared'; import { DATA_TOPIC } from '@podman/shared'; import { analyzeFrame } from '../vision/gemini.js'; import { detectCollisions } from '../collision/detector.js'; import { getGithubState } from '../github/client.js'; import { recordObservation, recordCollision, recordIntervention, updateInterventionStatus, } from '../memory/store.js'; import { getGitStates } from '../memory/db.js'; import { recallSimilar } from '../memory/vectors.js'; import { shouldIntervene, preferredAction } from '../memory/policy.js'; import { speak } from '../voice/live.js'; import { publishHermesMessage } from '../action/hermes.js'; export class PodMan { private contexts = new Map(); private encoder = new TextEncoder(); constructor( private room: Room, private podId: string, ) {} async start(): Promise { // Tier-2 optional ground-truth + engineer ACKs arrive over the data channel. this.room.on(RoomEvent.DataReceived, (payload) => { try { const msg = JSON.parse(new TextDecoder().decode(payload)) as DataMessage; if (msg.type === 'GIT_REPORT') { const c = this.contexts.get(msg.report.engineerId); if (c) c.hasUnpushedChanges = msg.report.unpushedCount > 0 || msg.report.dirtyFiles.length > 0; } if (msg.type === 'ACK') void updateInterventionStatus(msg.interventionId, msg.status); } catch { /* ignore malformed */ } }); } async onScreenFrame(engineerId: string, jpeg: Buffer): Promise { // Whole-frame guard: a Gemini/GitHub/Mongo failure on one frame must degrade // gracefully, never crash the long-running live agent loop. try { const ctx = await analyzeFrame(engineerId, this.podId, jpeg); this.contexts.set(engineerId, ctx); await recordObservation(ctx); // Fuse git ground truth: engineer_states written by scripts/podman-agent.mjs. // Keyed by name (matches --name arg), same as LiveKit participant identity. const gitStates = await getGitStates(this.podId); for (const [id, c] of this.contexts) { const git = gitStates.get(id); if (git && git.changedFiles.length > 0) c.hasUnpushedChanges = true; } const github = await getGithubState(); // cached const collisions = detectCollisions([...this.contexts.values()], github); for (const collision of collisions) await this.handle(collision); } catch (err) { console.warn(`[agent] frame from ${engineerId} skipped: ${(err as Error).message}`); } } private async handle(collision: Collision): Promise { const prior = await recallSimilar(collision); // Loop A: exact/vector recall raises confidence if (prior) collision.severity = 'critical'; if (!shouldIntervene(collision, prior)) return; // Loop B: policy gate await recordCollision(collision); const action = preferredAction(collision, prior); const names = collision.engineers.join(' and '); const message = `${names} are both editing ${collision.file}` + (collision.githubState?.unpushed ? ' and one has unpushed changes.' : '.') + (prior?.priorOutcome?.accepted ? ` I've seen this conflict pattern before; last time the team accepted the ${prior.priorIntervention?.suggestedAction.kind.replaceAll('_', ' ') ?? 'suggested'} action.` : prior ? ` I've seen this conflict pattern before.` : ''); const intervention: Intervention = { id: `int_${Date.now()}`, collisionId: collision.id, podId: this.podId, kind: 'card', message, suggestedAction: { kind: action, params: { file: collision.file, summary: message, engineers: collision.engineers, }, }, status: 'pending', createdAt: new Date().toISOString(), }; await recordIntervention(intervention); const data: DataMessage = { type: 'COLLISION', collision, intervention }; await this.room.localParticipant?.publishData(this.encoder.encode(JSON.stringify(data)), { reliable: true, topic: DATA_TOPIC, }); await publishHermesMessage(this.room, collision, intervention); if (collision.severity === 'critical') await speak(this.room, message); } }