1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
|
package main
// session.go bridges a phone browser to the Gemini Live API. The browser opens
// an app WebSocket at /ws/capture and exchanges small JSON envelopes:
//
// browser -> server:
// {"type":"audio","data":<b64 pcm16 16k>} mic chunk
// {"type":"video","data":<b64 jpeg>} camera frame (<=1 FPS)
// {"type":"text","text":"..."} typed message
// {"type":"photo","entity_id":N,"data":<b64 jpeg>,"caption":"..."} archival
//
// server -> browser:
// {"type":"ready"} Live session established
// {"type":"audio","data":<b64 pcm16 24k>} model speech to play
// {"type":"input_transcript","text":"..."} what the user said
// {"type":"output_transcript","text":"..."} what the model said
// {"type":"event","text":"..."} human-readable tool activity
// {"type":"turn_complete"}
// {"type":"interrupted"} stop/flush playback
// {"type":"error","text":"..."}
//
// The server owns the Gemini connection and the DB (tool runtime), so the API
// key never reaches the browser and tool calls are applied server-side.
import (
"context"
"encoding/base64"
"encoding/json"
"fmt"
"log"
"net/http"
"strconv"
"strings"
"sync"
"time"
"golang.org/x/net/websocket"
)
// browserMsg is the envelope the browser sends.
type browserMsg struct {
Type string `json:"type"`
Data string `json:"data,omitempty"` // base64
Text string `json:"text,omitempty"`
EntityID int64 `json:"entity_id,omitempty"`
Caption string `json:"caption,omitempty"`
}
// capturedPhoto is a finished photo-button result: the downscaled JPEG (base64),
// produced off-loop by the photo worker and fanned back into the capture loop
// for its two cheap side-effects (send to Gemini as a clientContent turn with
// the id as text; echo to the browser as a preview).
type capturedPhoto struct {
id string // the short id (e.g. "A3F9"), delivered to the model as text
b64 string // downscaled JPEG, base64
}
// capEvent is one fan-in item flowing into the single capture loop. Exactly one
// of browser/live/photo/err is set. browser/live come from the two socket
// readers; photo is a synthetic event the photo worker pushes after doing the
// heavy image work off-loop, so its socket writes happen on the loop like every
// other frame. A single goroutine owns all dispatch and state mutation — no
// locks needed across directions.
type capEvent struct {
browser *browserMsg // browser -> server frame
live *serverMessage // Gemini -> server frame
liveRaw []byte // raw bytes of the live frame (for goAway logging)
photo *capturedPhoto // finished photo-button result to push to live + browser
err error // non-nil: the source reader ended
fromLive bool // which reader produced this event
}
// evKind returns a short label naming the event being dispatched, for the
// slow-loop watchdog warning (e.g. "browser:audio", "live:toolCall").
func (ev capEvent) kind() string {
switch {
case ev.err != nil:
if ev.fromLive {
return "live:error"
}
return "browser:error"
case ev.photo != nil:
return "photo:captured"
case ev.browser != nil:
if ev.browser.Type != "" {
return "browser:" + ev.browser.Type
}
return "browser:?"
case ev.live != nil:
switch {
case ev.live.ToolCall != nil:
return "live:toolCall"
case ev.live.ServerContent != nil:
return "live:serverContent"
case ev.live.UsageMetadata != nil:
return "live:usage"
case ev.live.GoAway != nil:
return "live:goAway"
}
return "live:?"
}
return "?"
}
// browserWriter is the sole writer to the browser websocket. The capture loop
// and the offloaded photo goroutine hand it frames via enqueue; a dedicated
// goroutine drains the buffered channel and writes them, so no producer ever
// blocks on a slow browser socket. When the buffer is full we drop the OLDEST
// queued frame and keep the newest — realtime audio favours liveness over
// completeness. On a write error it cancels the session (the teardown watcher
// then closes both sockets). websocket.Conn permits concurrent method calls, so
// this writer coexists with the reader goroutine safely.
type browserWriter struct {
ws *websocket.Conn
ch chan []byte
cancel context.CancelFunc
mu sync.Mutex // serialises the drop-oldest drain across producers
}
// browserQueueSize bounds the outbound buffer. ~64 frames is a couple seconds of
// audio; beyond that we drop oldest rather than stall the loop.
const browserQueueSize = 64
// slowLoopThreshold is how long a single capture-loop dispatch may take before
// the watchdog logs a (rate-limited) warning. The loop is single-threaded, so
// anything slower than this stalls both directions.
const slowLoopThreshold = 10 * time.Millisecond
func newBrowserWriter(ws *websocket.Conn, cancel context.CancelFunc) *browserWriter {
return &browserWriter{ws: ws, ch: make(chan []byte, browserQueueSize), cancel: cancel}
}
// run drains the queue and writes to the socket until ctx is cancelled. A write
// error tears down the session. On cancellation it makes a best-effort flush of
// already-queued frames (e.g. a final "error" the handler enqueued right before
// returning) so they aren't lost to the shutdown race — but only until the
// socket write fails, since the teardown watcher will be closing ws concurrently.
func (b *browserWriter) run(ctx context.Context) {
for {
select {
case <-ctx.Done():
for {
select {
case msg := <-b.ch:
if err := websocket.Message.Send(b.ws, string(msg)); err != nil {
return
}
default:
return
}
}
case msg := <-b.ch:
if err := websocket.Message.Send(b.ws, string(msg)); err != nil {
b.cancel()
return
}
}
}
}
// enqueue marshals v and queues it for the browser. Non-blocking: if the buffer
// is full it drops the oldest queued frame, then enqueues v. Safe for concurrent
// producers (the loop and the photo goroutine).
func (b *browserWriter) enqueue(v map[string]any) {
msg, err := json.Marshal(v)
if err != nil {
return
}
b.mu.Lock()
defer b.mu.Unlock()
for {
select {
case b.ch <- msg:
return
default:
// Full: drop the oldest frame and retry. The drain may race an
// active writer draining too; either way we free a slot and loop.
select {
case <-b.ch:
default:
}
}
}
}
// captureSystemPrompt guides the model. Deliberately light-touch: it explains
// the tools and the open data model, and lets the model organise freely.
const captureSystemPrompt = `You are an inventory assistant helping the user record their belongings.
The user is opening boxes/containers and telling you (and showing you via camera) what is inside.
Your job: record what you observe into an inventory using the provided tools.
The data model is an open graph:
- "entities" have a free-text "kind" (e.g. "box", "item", "location", "tote") and a name.
- entities can carry arbitrary attributes (key/value), e.g. quantity, colour, brand, condition.
- entities can be related with free-text relations, e.g. an item is "contained_in" a box,
a box is "located_at" a location.
Guidelines:
- ALWAYS search before you create. Before recording any item, container or location, call
search first and reuse an existing entity if something similar already exists — only
record_entity when nothing matches. When in doubt, reuse.
- search takes structured fields, each a list of substrings OR-combined within the field,
with different fields AND-combined: kind, name, notes, attr_key, attr_value (or look up one
entity by id). Matching is case-insensitive substring. To find a place to put things in,
search by its kind, e.g. kind=["room","box","tote","shelf","location"], optionally narrowing
with name. Use several terms/synonyms/languages in one field, e.g. name=["cellar","keller"].
- When the user names a box or location, search for it first, then record it as an entity if
new, then relate items to it.
- To answer questions like "what's in/at X" (a room, box, container), search for X (e.g.
name=["X"]); each result lists its neighbours in relations_incoming (things located in /
inside it) and relations_outgoing (its own location/container). Answer from those, and
search by id to drill into a specific entity.
- Record each distinct item with a sensible name and useful attributes (quantity if known).
- The user can take a photo with the photo button. You then receive the photo along with
its id in text, like "photo id A3F9". If that photo shows an item, attach it by passing
image_id="A3F9" to record_entity (when first recording it) or update_entity, or use
attach_photo(entity_id, image_id) for an item you already recorded. You don't need the
camera still pointing at the item.
- If a photo would help identify or document an item, call request_photo.
- If the user says something was recorded by mistake or no longer exists, use delete_entity.
Prefer update_entity to fix details rather than deleting. Before deleting a box/container/
location that still holds items, warn the user out loud and only proceed (confirmed=true)
once they agree; the contained items are kept, only the containment links are removed.
- Be concise. Keep spoken replies short and natural — usually one short sentence.
Confirm what you recorded briefly, don't repeat back long lists, and ask clarifying
questions only when truly needed. Speak the same language the user speaks.
Primary speaker:
- Only one person is actually talking to you: the primary user. Identify them by
BOTH the initial, most consistent voice of the session AND the nearest/loudest
voice (the one holding the phone). Other voices — fainter, further away, or a
clearly different person (different pitch/timbre) — are background conversation
NOT addressed to you.
- Do NOT record, reply to, or run inventory tools for a voice that isn't the
primary user. Wait for the primary user instead.
- When you hear a clearly different voice and are therefore ignoring it, call
note_ignored_speaker with a short voice_characteristic (e.g. "deeper male
voice", "distant background voice") so the user can see you ignored it. Don't
spam it — call it once per distinct interruption, not for every buffer.`
// handleCaptureWS is the websocket.Handler for /ws/capture. Auth is enforced by
// the caller (session cookie checked before upgrade).
func (s *server) handleCaptureWS(ws *websocket.Conn) {
defer ws.Close()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// All browser-bound frames go through bw, a single async writer, so no
// producer ever blocks the capture loop on a slow browser socket.
bw := newBrowserWriter(ws, cancel)
go bw.run(ctx)
apiKey, err := geminiAPIKey()
if err != nil {
bw.enqueue(map[string]any{"type": "error", "text": "no API key: " + err.Error()})
return
}
live, err := dialLive(ctx, apiKey, s.model, captureSystemPrompt, inventoryTools)
if err != nil {
bw.enqueue(map[string]any{"type": "error", "text": "live connect: " + err.Error()})
return
}
// Teardown watcher: neither x/net/websocket read is context-aware, so the
// only way to unblock a reader parked in Receive() is to close its socket.
// This goroutine closes BOTH sockets once the context is cancelled, so a
// stop originating on either side (browser disconnect, Gemini EOF/goAway,
// read-deadline, or any handler returning) tears down the whole session
// symmetrically. cancel() (deferred above) is what always fires it.
go func() {
<-ctx.Done()
dbg("capture: ctx cancelled -> tearing down (closing Gemini + browser sockets)")
live.close()
ws.Close()
}()
// Record a capture session row for audit.
res := mustExec(s.db, `INSERT INTO capture_sessions (started_at) VALUES (?)`, time.Now().Unix())
sessionRowID, _ := res.LastInsertId()
// Stats: per-session counters, a server-side silence gate, and an SSE hub
// the capture page subscribes to via a short-lived correlation token.
stats := newSessionStats()
// On teardown, stamp ended_at and persist the final token/cost totals. The
// per-reply persist below keeps the row current throughout, so a crash still
// leaves accurate data; this final write just captures the last usage frame
// and the end time.
defer func() {
s.persistSessionUsage(sessionRowID, stats)
mustExec(s.db, `UPDATE capture_sessions SET ended_at=? WHERE id=?`, time.Now().Unix(), sessionRowID)
}()
gate := newSilenceGate(defaultGate, activeModel.AudioInSampleRate)
pending := newPendingPhotos(25)
hub := newStatsHub()
statsToken := randomBase64(18)
s.statsReg.register(statsToken, hub)
defer s.statsReg.unregister(statsToken)
// Tell the browser the token so it can open the SSE stats stream.
bw.enqueue(map[string]any{"type": "stats_token", "token": statsToken})
bw.enqueue(map[string]any{"type": "ready"})
// Fan-in channel: both readers funnel frames here so a single loop below
// owns all dispatch and state mutation. Unbuffered is fine — the loop is the
// only consumer and never blocks on a slow external call while holding state.
events := make(chan capEvent)
// Reader goroutine: browser -> events. Closing ws (via the watcher) unblocks
// Receive and makes this exit with an error event.
go func() {
for {
var raw []byte
if err := websocket.Message.Receive(ws, &raw); err != nil {
dbg("browser: receive ended (disconnect/closed): %v", err)
sendEvent(ctx, events, capEvent{err: err, fromLive: false})
return
}
var m browserMsg
if err := json.Unmarshal(raw, &m); err != nil {
continue
}
// Audio is high-frequency and mostly silence; it has its own
// gate-based logging (see "-> audio in start/end") once it is actually
// forwarded to the live socket. Log the other, rarer frame types here.
if m.Type != "audio" {
dbg("browser -> %s (%d b64 chars)", m.Type, len(m.Data))
}
if !sendEvent(ctx, events, capEvent{browser: &m}) {
return
}
}
}()
// Reader goroutine: Gemini -> events. Closing live (via the watcher) or the
// read-deadline unblocks receive and makes this exit with an error event.
go func() {
for {
msg, raw, err := live.receive()
if err != nil {
dbg("live: receive ended (EOF/read-deadline/closed): %v", err)
sendEvent(ctx, events, capEvent{err: err, fromLive: true})
return
}
if msg == nil {
continue
}
if !sendEvent(ctx, events, capEvent{live: msg, liveRaw: raw, fromLive: true}) {
return
}
}
}()
// Periodic stats push to SSE, driven off the same context.
statsTick := time.NewTicker(1 * time.Second)
defer statsTick.Stop()
// audioOutStarted marks whether we've logged the start of the current turn's
// model audio; reset at each turn boundary so we log one "audio out start".
audioOutStarted := false
// lastSlowWarn rate-limits the slow-dispatch watchdog to ~1 warning/second so
// a chronically slow session doesn't flood stderr. Loop-owned; no lock.
var lastSlowWarn time.Time
// Single capture loop: dispatches browser frames, Gemini frames, and the
// stats tick. Any reader error, or ctx cancellation, ends the session; the
// deferred cancel() then fires the watcher that closes both sockets.
//
// Everything below runs inline on this one goroutine, so a slow handler
// stalls all dispatch. Browser writes are async (bw) and photo downscaling is
// offloaded; the remaining inline slow path is tool-call/DB dispatch, which
// the watchdog below measures. We time only the dispatch (not the idle select
// wait) and warn when it exceeds slowLoopThreshold, naming the event kind.
for {
select {
case <-ctx.Done():
return
case <-statsTick.C:
hub.send("stats", stats.snapshot())
case ev := <-events:
t0 := time.Now()
if ev.err != nil {
// A reader stopped: the session is over. Returning triggers the
// deferred cancel(), which closes the other socket too.
if ev.fromLive {
select {
case <-ctx.Done():
default:
bw.enqueue(map[string]any{"type": "error", "text": "live closed: " + ev.err.Error()})
}
}
return
}
if ev.browser != nil {
if !s.handleBrowserFrame(bw, live, ctx, events, *ev.browser, gate, stats, hub, pending) {
return
}
}
if ev.live != nil {
s.handleLiveFrame(bw, live, ev.live, ev.liveRaw, stats, hub, pending, &audioOutStarted, sessionRowID)
}
if ev.photo != nil {
// A downscaled photo finished off-loop. Send it to Gemini as a
// clientContent turn (turnComplete=false) so the image AND its id
// persist in context in order — a realtimeInput video frame would be
// overwritten by the next camera frame within milliseconds. The id
// travels as text, so attaching never depends on reading it off the
// image. A send failure means the socket is dead, so end the session
// (like audio).
parts := []part{
{InlineData: &blob{Data: ev.photo.b64, MimeType: "image/jpeg"}},
{Text: fmt.Sprintf("[system] The user just captured an archival photo with id %s. "+
"To attach it to an item, pass image_id=%q to record_entity/update_entity, "+
"or call attach_photo(entity_id, image_id=%q).",
ev.photo.id, ev.photo.id, ev.photo.id)},
}
if err := live.sendClientContent(parts, false); err != nil {
dbg("photo %s: sendClientContent failed (live socket dead?): %v", ev.photo.id, err)
return
}
bw.enqueue(map[string]any{"type": "photo_preview", "id": ev.photo.id, "data": ev.photo.b64})
dbg("photo %s: sent as clientContent (image+id text) + preview to browser", ev.photo.id)
}
if d := time.Since(t0); d > slowLoopThreshold {
now := time.Now()
if now.Sub(lastSlowWarn) > time.Second {
lastSlowWarn = now
log.Printf("capture: slow loop dispatch %s for %s", d.Round(time.Millisecond), ev.kind())
}
}
}
}
}
// handleBrowserFrame processes one browser -> server envelope, forwarding to
// Gemini and updating stats. It returns false if the session should end (a send
// to Gemini failed), which makes the capture loop return and tear down.
func (s *server) handleBrowserFrame(bw *browserWriter, live *liveConn, ctx context.Context, events chan capEvent, m browserMsg, gate *silenceGate, stats *sessionStats, hub *statsHub, pending *pendingPhotos) bool {
switch m.Type {
case "audio":
if m.Data == "" {
return true
}
// Server-side silence gate: only forward (and bill) speech. We run manual
// VAD (Gemini's auto-VAD is disabled), so we bracket each speech turn with
// activityStart (before the buffers) and activityEnd (after them).
dec := gate.push(m.Data, time.Now())
if dec.stateChanged && dec.open {
// Speech onset: open the turn BEFORE sending its audio (incl. preroll).
if err := live.sendActivityStart(); err != nil {
return false
}
stats.setGate(true)
hub.send("audio_state", map[string]any{"on": true})
dbg("-> activityStart (speech), forwarding %d buffers", len(dec.forward))
}
for _, b64 := range dec.forward {
if err := live.sendAudio(b64); err != nil {
return false
}
if raw, e := base64.StdEncoding.DecodeString(b64); e == nil {
stats.addAudioIn(len(raw))
}
}
if dec.stateChanged && !dec.open {
// Speech end: close the turn AFTER its last buffer so the model replies.
if err := live.sendActivityEnd(); err != nil {
return false
}
stats.setGate(false)
hub.send("audio_state", map[string]any{"on": false})
dbg("-> activityEnd (silence)")
}
case "video":
if m.Data != "" {
if err := live.sendVideoFrame(m.Data); err != nil {
dbg("video frame -> live failed (socket dead?): %v", err)
return false
}
stats.addVideoFrame()
}
case "text":
if m.Text != "" {
if err := live.sendText(m.Text); err != nil {
dbg("text -> live failed (socket dead?): %v", err)
return false
}
}
case "photo":
s.storeArchivalPhoto(bw, m)
case "photo_button":
s.handlePhotoButton(bw, ctx, events, pending, m)
case "client_error":
// An uncaught JS exception from the browser, forwarded so mobile-only
// failures surface in the journal. Always logged (not just dbg).
log.Printf("capture: client error: %s", m.Text)
}
return true
}
// sendEvent delivers one fan-in event, but gives up if the context is cancelled
// (the loop has stopped consuming) so a reader goroutine can never block forever
// on the channel during teardown. Returns false if the send was abandoned.
func sendEvent(ctx context.Context, ch chan<- capEvent, ev capEvent) bool {
select {
case ch <- ev:
return true
case <-ctx.Done():
return false
}
}
// handlePhotoButton is the photo worker: it stores the full-res photo in pending
// under a minted id and produces a downscaled copy to send to the model. The id
// is delivered to the model as text (in the clientContent turn), letting it
// attach the archival photo later via the image_id tool argument.
//
// The heavy work (base64 decode, downscale, JPEG re-encode) is tens–hundreds of
// ms, so it runs in its own goroutine rather than on the capture loop, where it
// would stall model-audio playback. When done it pushes a synthetic
// capEvent{photo:…} back onto the fan-in channel; the loop then does the two
// cheap side-effects (send to Gemini, echo preview) on the loop goroutine,
// keeping all socket writes serialised there. sendEvent drops the result cleanly
// if the session is tearing down.
//
// Safe without extra locks: pendingPhotos is mutex-guarded and sendEvent/events
// is a channel; the worker no longer touches live or bw directly.
func (s *server) handlePhotoButton(bw *browserWriter, ctx context.Context, events chan capEvent, pending *pendingPhotos, m browserMsg) {
if m.Data == "" {
dbg("photo_button: empty data, ignoring")
return
}
go func() {
fullres, err := base64.StdEncoding.DecodeString(m.Data)
if err != nil {
dbg("photo_button: base64 decode failed: %v", err)
return
}
id := pending.add(fullres)
t0 := time.Now()
small, err := makeDownscaledFrame(fullres)
if err != nil {
dbg("photo %s: makeDownscaledFrame failed: %v", id, err)
bw.enqueue(map[string]any{"type": "error", "text": "photo: " + err.Error()})
return
}
smallB64 := base64.StdEncoding.EncodeToString(small)
dbg("photo %s: full-res %d bytes -> downscaled %d bytes in %s",
id, len(fullres), len(small), time.Since(t0).Round(time.Millisecond))
// Hand the finished frame to the loop for its socket writes.
if !sendEvent(ctx, events, capEvent{photo: &capturedPhoto{id: id, b64: smallB64}}) {
dbg("photo %s: session tearing down, dropped", id)
}
}()
}
// handleCaptureStats streams session cost/usage stats to the capture page over
// Server-Sent Events. Correlated to the capture WS session by ?token=. Auth via
// the session cookie.
func (s *server) handleCaptureStats(w http.ResponseWriter, r *http.Request) {
if _, _, ok := currentSession(s.db, w, r); !ok {
http.Error(w, "unauthorized", http.StatusUnauthorized)
return
}
token := r.URL.Query().Get("token")
hub, ok := s.statsReg.get(token)
if !ok {
http.Error(w, "unknown or expired stats token", http.StatusNotFound)
return
}
flusher, ok := w.(http.Flusher)
if !ok {
http.Error(w, "streaming unsupported", http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
w.Header().Set("X-Accel-Buffering", "no")
w.WriteHeader(http.StatusOK)
flusher.Flush()
for {
select {
case <-r.Context().Done():
return
case <-hub.closed:
return
case msg := <-hub.ch:
if _, err := w.Write(msg); err != nil {
return
}
flusher.Flush()
}
}
}
// handleLiveFrame processes one Gemini -> server message: forwarding audio,
// transcripts and tool activity to the browser, and dispatching tool calls
// against the DB. It runs on the single capture loop goroutine, so it needs no
// locking around stats/pending. audioOutStarted is owned by the loop and points
// at its per-turn "logged audio start" flag.
func (s *server) handleLiveFrame(bw *browserWriter, live *liveConn, msg *serverMessage, raw []byte, stats *sessionStats, hub *statsHub, pending *pendingPhotos, audioOutStarted *bool, sessionRowID int64) {
if msg.UsageMetadata != nil {
stats.applyUsage(msg.UsageMetadata)
hub.send("stats", stats.snapshot())
// Persist the running token/cost totals so the row is always current,
// even if the session ends abruptly (crash, hard disconnect).
s.persistSessionUsage(sessionRowID, stats)
}
if sc := msg.ServerContent; sc != nil {
if sc.Interrupted {
dbg("<- interrupted")
bw.enqueue(map[string]any{"type": "interrupted"})
}
if sc.InputTranscription != nil && sc.InputTranscription.Text != "" {
dbg("<- input_transcript %q", sc.InputTranscription.Text)
bw.enqueue(map[string]any{"type": "input_transcript", "text": sc.InputTranscription.Text})
}
if sc.OutputTranscription != nil && sc.OutputTranscription.Text != "" {
dbg("<- output_transcript %q", sc.OutputTranscription.Text)
bw.enqueue(map[string]any{"type": "output_transcript", "text": sc.OutputTranscription.Text})
}
if sc.ModelTurn != nil {
for _, p := range sc.ModelTurn.Parts {
if p.InlineData != nil && p.InlineData.Data != "" {
if !*audioOutStarted {
dbg("<- audio out start")
*audioOutStarted = true
}
bw.enqueue(map[string]any{"type": "audio", "data": p.InlineData.Data})
if b, e := base64.StdEncoding.DecodeString(p.InlineData.Data); e == nil {
stats.addAudioOut(len(b))
}
}
}
}
if sc.TurnComplete {
dbg("<- turn_complete")
*audioOutStarted = false
bw.enqueue(map[string]any{"type": "turn_complete"})
}
}
if tc := msg.ToolCall; tc != nil {
t0 := time.Now()
names := make([]string, len(tc.FunctionCalls))
for i, c := range tc.FunctionCalls {
names[i] = c.Name
}
dbg("<- toolCall %v", names)
s.handleToolCalls(bw, live, tc.FunctionCalls, pending)
dbg(" toolCall handled in %s", time.Since(t0).Round(time.Millisecond))
}
if msg.GoAway != nil {
bw.enqueue(map[string]any{"type": "error", "text": "server going away"})
log.Printf("capture: goAway: %s", raw)
// Proactively end the session: closing the Gemini socket makes its
// reader exit and the loop tear everything down.
live.close()
}
}
// persistSessionUsage writes the current token/cost totals for a session into
// its capture_sessions row. Called on every usageMetadata reply (so the row is
// continuously current) and once more at teardown. Cost is stored as computed
// now, freezing it against future rate changes; the raw token counts are stored
// too so it can be recomputed if ever needed.
func (s *server) persistSessionUsage(sessionRowID int64, stats *sessionStats) {
u := stats.finalUsage()
mustExec(s.db, `UPDATE capture_sessions SET
model=?,
tokens_audio_in=?, tokens_audio_out=?, tokens_image_in=?,
tokens_text_in=?, tokens_text_out=?, tokens_thoughts=?,
tokens_tool_use=?, tokens_cached=?, tokens_total=?,
video_frames=?, cost_usd=?
WHERE id=?`,
s.model,
u.AudioInTok, u.AudioOutTok, u.ImageInTok,
u.TextInTok, u.TextOutTok, u.ThoughtsTok,
u.ToolUseTok, u.CachedTok, u.TotalTok,
u.VideoFrames, round4(u.CostTotal),
sessionRowID)
}
// handleToolCalls executes each function call against the DB and returns the
// results to Gemini, while surfacing a human-readable event to the browser for
// EVERY tool call. Tools may return a tailored event string; when they don't
// (e.g. search, or a validation/error path), we synthesize a generic one from
// the call name and result so no tool activity is ever invisible in the UI.
// Events whose result carries an "error" key are flagged level=error so the
// frontend renders them in red.
func (s *server) handleToolCalls(bw *browserWriter, live *liveConn, calls []functionCall, pending *pendingPhotos) {
resps := make([]functionResponse, 0, len(calls))
for _, call := range calls {
result, event, browserAction := s.dispatchInventoryTool(call, pending)
resps = append(resps, functionResponse{ID: call.ID, Name: call.Name, Response: result})
_, isErr := result["error"]
if event == "" {
event = fallbackToolEvent(call, result, isErr)
}
msg := map[string]any{"type": "event", "text": event}
if isErr {
msg["level"] = "error"
}
bw.enqueue(msg)
if browserAction != nil {
bw.enqueue(browserAction)
}
}
if err := live.sendToolResponses(resps); err != nil {
log.Printf("capture: send tool responses: %v", err)
}
}
// fallbackToolEvent builds a generic human-readable line for a tool call that
// didn't provide its own event text (e.g. search, or an error/validation path).
// It gives the user visibility into every tool the model invokes.
func fallbackToolEvent(call functionCall, result map[string]any, isErr bool) string {
if isErr {
if e, ok := result["error"].(string); ok && e != "" {
return fmt.Sprintf("⚠️ %s: %s", toolLabel(call.Name), e)
}
return fmt.Sprintf("⚠️ %s failed", toolLabel(call.Name))
}
switch call.Name {
case "search":
q := ""
if v, ok := call.Args["query"].(string); ok && v != "" {
q = " " + strconv.Quote(v)
} else if id := argInt(call.Args, "id"); id != 0 {
q = fmt.Sprintf(" #%d", id)
}
n := 0
if c, ok := result["count"].(int); ok {
n = c
}
return fmt.Sprintf("🔍 searched%s (%d result%s)", q, n, plural(n))
default:
return "🔧 " + toolLabel(call.Name)
}
}
// toolLabel turns a tool name like "record_entity" into "record entity".
func toolLabel(name string) string {
if name == "" {
return "tool"
}
return strings.ReplaceAll(name, "_", " ")
}
func plural(n int) string {
if n == 1 {
return ""
}
return "s"
}
// storeArchivalPhoto persists a full-res JPEG from the browser as a blob.
func (s *server) storeArchivalPhoto(bw *browserWriter, m browserMsg) {
if m.EntityID == 0 || m.Data == "" {
return
}
data, err := base64.StdEncoding.DecodeString(m.Data)
if err != nil {
return
}
var exists int
s.db.QueryRow(`SELECT COUNT(*) FROM entities WHERE id=?`, m.EntityID).Scan(&exists)
if exists == 0 {
bw.enqueue(map[string]any{"type": "error", "text": "photo: unknown entity"})
return
}
mustExec(s.db, `INSERT INTO photos (entity_id, blob, mime, caption, created_at) VALUES (?,?,?,?,?)`,
m.EntityID, data, "image/jpeg", m.Caption, time.Now().Unix())
bw.enqueue(map[string]any{"type": "event", "text": "📷 photo saved"})
}
// authedCaptureWS is the HTTP entry point: it checks the session cookie, then
// hands off to the websocket handler.
func (s *server) authedCaptureWS(w http.ResponseWriter, r *http.Request) {
if _, _, ok := currentSession(s.db, w, r); !ok {
http.Error(w, "unauthorized", http.StatusUnauthorized)
return
}
websocket.Handler(s.handleCaptureWS).ServeHTTP(w, r)
}
|