diff --git a/cmd/mavend/clarify_test.go b/cmd/mavend/clarify_test.go index 425a203..16a2807 100644 --- a/cmd/mavend/clarify_test.go +++ b/cmd/mavend/clarify_test.go @@ -439,7 +439,7 @@ func TestClarifySecondGapExhaustionResumesLowerFlow(t *testing.T) { h.clarifyStore.Push(voiceDialogueID, older) h.clarifyStore.Push(voiceDialogueID, top) - reply := h.runTurn(ctx, "купить хлеб", sourceText) + reply := h.runTurn(ctx, router.NormalizedInput{Text: "купить хлеб", Source: sourceText}) resumed, _ := clarifyResumedFor(dialogue.SlotTime) want := withResumed(clarifyGaveUp, resumed) if reply != want { diff --git a/cmd/mavend/reactive_notes_test.go b/cmd/mavend/reactive_notes_test.go index 61cf5c4..d72c459 100644 --- a/cmd/mavend/reactive_notes_test.go +++ b/cmd/mavend/reactive_notes_test.go @@ -126,7 +126,7 @@ func TestRunTurnExplicitNoteStoresOnlyTheBody(t *testing.T) { } const utterance = "запомни: запасной ключ лежит в синей коробке" - if reply := h.runTurn(ctx, utterance, sourceText); reply != "сохранила заметку." { + if reply := h.runTurn(ctx, router.NormalizedInput{Text: utterance, Source: sourceText}); reply != "сохранила заметку." { t.Fatalf("reply = %q, want the fixed feminine acknowledgement", reply) } if model.calls != 0 { diff --git a/cmd/mavend/reminder_cancel_test.go b/cmd/mavend/reminder_cancel_test.go index 68d992c..727a3f3 100644 --- a/cmd/mavend/reminder_cancel_test.go +++ b/cmd/mavend/reminder_cancel_test.go @@ -382,7 +382,7 @@ func TestReminderCancellationIsAPreRouteTurnAndDoesNotGetSwallowedByClarify(t *t Utterance: "напомни позвонить маме", Asked: h.now(), TTL: clarifyTTL, }) - reply := h.runTurn(ctx, "отмени напоминание про врача", sourceText) + reply := h.runTurn(ctx, router.NormalizedInput{Text: "отмени напоминание про врача", Source: sourceText}) if !strings.Contains(reply, clarifyDropped) || !strings.Contains(reply, "отменила напоминание") { t.Fatalf("reply = %q, want dropped clarify notice and cancellation", reply) } diff --git a/cmd/mavend/repair_test.go b/cmd/mavend/repair_test.go index a030a4a..4dcdc2c 100644 --- a/cmd/mavend/repair_test.go +++ b/cmd/mavend/repair_test.go @@ -149,7 +149,7 @@ func TestRepairResumesQuestionParkedAfterTheCorrectedTurn(t *testing.T) { t.Fatal("expected a parked reminder question") } - reply := h.runTurn(ctx, "нет, это был вопрос", sourceText) + reply := h.runTurn(ctx, router.NormalizedInput{Text: "нет, это был вопрос", Source: sourceText}) resumed, _ := clarifyResumedFor(dialogue.SlotTime) if !strings.HasSuffix(reply, resumed) { t.Fatalf("the correction hid the still-live question: reply=%q want suffix=%q", reply, resumed) @@ -176,14 +176,14 @@ func TestRepairedClarifyCompletesWithoutDroppingTheOlderQuestion(t *testing.T) { t.Fatal("expected the older reminder question") } - if reply := h.runTurn(ctx, "нет, это было напоминание", sourceText); !strings.Contains(reply, "Когда") { + if reply := h.runTurn(ctx, router.NormalizedInput{Text: "нет, это было напоминание", Source: sourceText}); !strings.Contains(reply, "Когда") { t.Fatalf("the repaired reminder did not ask for its missing time: %q", reply) } if depth := h.clarifyStore.Depth(voiceDialogueID); depth != 2 { t.Fatalf("the repaired question overwrote the older one: depth=%d want=2", depth) } - reply := h.runTurn(ctx, "сегодня в 15:00", sourceText) + reply := h.runTurn(ctx, router.NormalizedInput{Text: "сегодня в 15:00", Source: sourceText}) resumed, _ := clarifyResumedFor(dialogue.SlotTime) if !strings.HasSuffix(reply, resumed) { t.Fatalf("completing the repaired request did not resume the older one: reply=%q", reply) diff --git a/cmd/mavend/routingtrace.go b/cmd/mavend/routingtrace.go index 6f529ed..c418132 100644 --- a/cmd/mavend/routingtrace.go +++ b/cmd/mavend/routingtrace.go @@ -120,6 +120,7 @@ func (h *reactiveHandler) persistDecision(turnCtx context.Context, rec *decision Source: string(src), Winner: rec.Winner, Intent: wonIntent(rec), + RouteProducer: rec.RouteProducer, ClaimedBeforeHead: claimedBeforeHead(rec), EncoderID: h.encoderID, Outcome: wonAt(rec, decision.StageAction), diff --git a/cmd/mavend/simulator_test.go b/cmd/mavend/simulator_test.go index bc4d39c..fb65845 100644 --- a/cmd/mavend/simulator_test.go +++ b/cmd/mavend/simulator_test.go @@ -696,7 +696,7 @@ func (w *simWorld) stimulate(ctx context.Context, s step) { switch { case s.Say != "": - reply := w.handler.runTurn(ctx, s.Say, sourceText) + reply := w.handler.runTurn(ctx, router.NormalizedInput{Text: s.Say, Source: sourceText}) w.replies = append(w.replies, reply) w.logf("он: %s", s.Say) w.logf("она: %s", reply) diff --git a/cmd/mavend/turnrole_test.go b/cmd/mavend/turnrole_test.go index 767b0ae..ea4df74 100644 --- a/cmd/mavend/turnrole_test.go +++ b/cmd/mavend/turnrole_test.go @@ -264,10 +264,10 @@ func TestClarifyCancelEndsTheExchange(t *testing.T) { // the same memo. func TestTheTurnIsRoutedOnce(t *testing.T) { h, _ := newRoutingClarifyHandler(t) - rt := h.newTurnRoute("какая сейчас погода в Риме?", h.now()) + rt := h.newTurnRoute(router.NormalizedInput{Text: "какая сейчас погода в Риме?", Source: sourceText}, h.now()) ctx := withTurnRoute(withDialogueID(context.Background(), voiceDialogueID), rt) - first, ok := h.routeForRole(ctx, rt.text) + first, ok := h.routeForRole(ctx, rt.input.Text) if !ok { t.Fatal("the cascade must produce a decision to classify against") } diff --git a/cmd/mavend/turnroute.go b/cmd/mavend/turnroute.go index ad68af8..185b402 100644 --- a/cmd/mavend/turnroute.go +++ b/cmd/mavend/turnroute.go @@ -18,9 +18,9 @@ import ( // second on the resident model and — worse — could disagree with itself, which // is exactly the class of bug this task is about. type turnRoute struct { - h *reactiveHandler - text string - now time.Time + h *reactiveHandler + input router.NormalizedInput + now time.Time once sync.Once dec router.Decision @@ -46,8 +46,8 @@ type turnRoute struct { type turnRouteKey struct{} -func (h *reactiveHandler) newTurnRoute(text string, now time.Time) *turnRoute { - return &turnRoute{h: h, text: text, now: now} +func (h *reactiveHandler) newTurnRoute(input router.NormalizedInput, now time.Time) *turnRoute { + return &turnRoute{h: h, input: input, now: now} } func withTurnRoute(ctx context.Context, rt *turnRoute) context.Context { @@ -70,7 +70,7 @@ func (r *turnRoute) resolve(ctx context.Context) (router.Decision, bool, *dialog if r.h.dialogueSessions != nil { r.prev = r.h.dialogueSessions.Get(dialogueIDOf(ctx), r.now) } - if dec, cont := continuationDecision(r.prev, r.text, r.now); cont { + if dec, cont := continuationDecision(r.prev, r.input.Text, r.now); cont { log.Printf("voice: continuation of %s from the previous turn", dec.Intent) r.dec, r.cont = dec, true return @@ -79,7 +79,7 @@ func (r *turnRoute) resolve(ctx context.Context) (router.Decision, bool, *dialog r.err = router.ErrNoIntents return } - r.dec, r.err = r.h.router.Route(ctx, r.text, r.now) + r.dec, r.err = r.h.router.Route(ctx, r.input.Text, r.now) }) return r.dec, r.cont, r.prev, r.err } @@ -92,7 +92,7 @@ func (r *turnRoute) resolve(ctx context.Context) (router.Decision, bool, *dialog func (h *reactiveHandler) routeForRole(ctx context.Context, text string) (router.Decision, bool) { rt := turnRouteFrom(ctx) if rt == nil { - rt = h.newTurnRoute(text, h.now()) + rt = h.newTurnRoute(router.NormalizedInput{Text: text, Source: sourceText}, h.now()) } dec, _, _, err := rt.resolve(ctx) if err != nil { diff --git a/cmd/mavend/voice.go b/cmd/mavend/voice.go index 1f157f9..b603709 100644 --- a/cmd/mavend/voice.go +++ b/cmd/mavend/voice.go @@ -211,7 +211,7 @@ func (h *reactiveHandler) HandlePushToTalk(ctx context.Context, req voice.PushTo // 2-5. the shared turn pipeline (confirm → clarify → route → dialogue → // action → replier), identical to the text path. - replyText := h.runTurn(ctx, text, sourceVoice) + replyText := h.runTurn(ctx, router.NormalizedInput{Text: text, Source: sourceVoice}) // 6. tts — synthesise the reply text; return to the voice server which // ships it back on the conn. @@ -244,30 +244,30 @@ func (h *reactiveHandler) upgradeAPI(api ipc.CoreAPI) { // HandlePushToTalk so text channels share the same routing logic. func (h *reactiveHandler) handleText(ctx context.Context, conversation, text string) string { log.Printf("voice: handleText: %q", text) - return h.runTurn(withDialogueID(ctx, dialogueIDFor(sourceText, conversation)), text, sourceText) + return h.runTurn(withDialogueID(ctx, dialogueIDFor(sourceText, conversation)), router.NormalizedInput{Text: text, Source: sourceText}) } -// turnSource — which channel this utterance arrived on, in the same provenance -// vocabulary facts use (internal/event). It is threaded through runTurn because -// a turn can write a fact, and a fact that lies about where it came from is -// worse than no fact: provenance is the first column read when asking why a -// daemon-wide setting is the way it is. -type turnSource string +// turnSource is a local alias for router.InputSource, kept so the daemon code +// reads sourceVoice/sourceText without a package prefix at every call site. +// The canonical type lives in the router package; this is pure convenience. +type turnSource = router.InputSource const ( - sourceVoice turnSource = "tap:voice" // HandlePushToTalk, a real microphone - sourceText turnSource = "tap:text" // handleText: mavweb /api/chat, telegram + sourceVoice = router.InputSourceVoice + sourceText = router.InputSourceText ) // runTurn — the reactive turn pipeline shared by the voice and text entry // points: expired-clarify notice → confirm answer → explicit correction → // clarify answer → quiet toggle → reminder cancellation → route → dialogue // merge → clarify question → action → replier. -// Takes the already-transcribed utterance, returns the reply text; the voice -// path wraps it in stt/tts, the text path returns it as-is. +// Takes the NormalizedInput (typed ingress boundary), returns the reply text; +// the voice path wraps it in stt/tts, the text path returns it as-is. // // The ordering is load-bearing — see the step comments. -func (h *reactiveHandler) runTurn(ctx context.Context, text string, src turnSource) (reply string) { +func (h *reactiveHandler) runTurn(ctx context.Context, input router.NormalizedInput) (reply string) { + text := input.Text + src := input.Source // 0. the decision record (V-564). Installed here rather than in the IPC // entry point, so the mic, telegram and the web all leave the same trail — // a record only the web produced would be missing exactly the turns that @@ -275,7 +275,7 @@ func (h *reactiveHandler) runTurn(ctx context.Context, text string, src turnSour // on a human-rate path, and no claim site can change a route with it. if h.decisions != nil { var rec *decision.Record - ctx, rec = decision.With(ctx, text) + ctx, rec = decision.With(ctx, text, string(src)) decision.Expect(ctx, decision.StagePreRoute, preRouteLadder) defer func() { done := rec.Finish(h.now()) @@ -289,7 +289,7 @@ func (h *reactiveHandler) runTurn(ctx context.Context, text string, src turnSour // claiming it, and step 5 acts on the same decision — routing twice would // cost a second on the resident model and could disagree with itself. now := h.now() - rt := h.newTurnRoute(text, now) + rt := h.newTurnRoute(input, now) ctx = withTurnRoute(ctx, rt) // A resolver may suspend an older clarify flow even when it handles this // turn itself. Finalise that state at one choke point so early returns from @@ -414,6 +414,10 @@ func (h *reactiveHandler) runTurn(ctx context.Context, text string, src turnSour return withNotice(expiredNotice, "не получилось разобрать команду.") } log.Printf("voice: route result: intent=%s slots=%+v", dec.Intent, dec.Slots) + // Carry the route producer into the decision record for observability. + if rec := decision.From(ctx); rec != nil && dec.Producer != "" { + rec.RouteProducer = string(dec.Producer) + } // 7. dialogue — fill this turn's missing slots from a prior same-intent // turn (follow-ups like «напомни завтра» → «…позвонить маме»), then remember diff --git a/internal/decision/decision.go b/internal/decision/decision.go index 25afb63..375532c 100644 --- a/internal/decision/decision.go +++ b/internal/decision/decision.go @@ -68,6 +68,16 @@ type Record struct { Winner string `json:"winner"` Claims []Claim `json:"claims"` + // InputSource — which channel the utterance arrived on (tap:voice or + // tap:text). Carried for observability so a trace can distinguish a voice + // turn from a text turn without re-deriving it from surrounding claims. + InputSource string `json:"input_source,omitempty"` + + // RouteProducer — which cascade stage produced the routing decision. + // Carried for observability so a trace names the winning component directly + // rather than requiring a scan of the claims list. + RouteProducer string `json:"route_producer,omitempty"` + mu sync.Mutex rosters []roster } @@ -152,9 +162,10 @@ func (r *Record) Finish(now time.Time) *Record { type recorderKey struct{} // With returns a context carrying a fresh record, and the record to read after -// the turn has answered. -func With(ctx context.Context, utterance string) (context.Context, *Record) { - rec := &Record{Utterance: utterance} +// the turn has answered. inputSource identifies the channel the utterance +// arrived on; pass empty if unknown. +func With(ctx context.Context, utterance string, inputSource string) (context.Context, *Record) { + rec := &Record{Utterance: utterance, InputSource: inputSource} return context.WithValue(ctx, recorderKey{}, rec), rec } diff --git a/internal/router/intent.go b/internal/router/intent.go index 3cbac14..a538fe2 100644 --- a/internal/router/intent.go +++ b/internal/router/intent.go @@ -35,6 +35,18 @@ package router import "time" +// RouteProducer — which stage of the cascade produced the routing decision. +// Recorded for observability so a trace can name the winning component without +// re-deriving it from the stage number and surrounding claims. +type RouteProducer string + +const ( + RouteProducerGrammar RouteProducer = "grammar" + RouteProducerHeads RouteProducer = "heads" + RouteProducerLLM RouteProducer = "llm" + RouteProducerClassifier RouteProducer = "classifier" +) + // Intent — the seven save-where labels from docs/design.md's routing table. The // discriminator is "does the loop evaluate a predicate against it?": // @@ -105,6 +117,10 @@ type Decision struct { Slots Slots Clarify bool // stage 3: below threshold — ask, don't guess + // Producer — which cascade stage produced this decision. Recorded for + // observability so a trace can name the winning component directly. + Producer RouteProducer + // Source — where the answer lives, for a query. The second half of the // route, and empty on every other intent. SourceUnknown means no decider // named one and the daemon walks its whole chain, which is what shipped diff --git a/internal/router/router.go b/internal/router/router.go index 94d9bfa..d7e5374 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -97,6 +97,7 @@ func (r *Router) Route(ctx context.Context, utterance string, now time.Time) (De continue // grammar matched shape but not content → fall through } d.Utterance = utterance + d.Producer = RouteProducerGrammar // A literal pattern named that destination, which is the one provenance // allowed to take the personal boundary off a turn (V-666). Set here and // nowhere else, so no other arm of the cascade can claim it. @@ -138,6 +139,7 @@ func (r *Router) Route(ctx context.Context, utterance string, now time.Time) (De Confidence: res.Confidence, Source: res.Source, Clarify: res.Clarify, + Producer: RouteProducerHeads, } r.fillSlots(ctx, &d, now) // The clarify head relearned the English assumption that one word @@ -180,6 +182,7 @@ func (r *Router) Route(ctx context.Context, utterance string, now time.Time) (De if r.llm != nil { if d, ok, err := r.llm.Route(ctx, utterance, now); err == nil && ok { d.Utterance = utterance + d.Producer = RouteProducerLLM r.fillSlots(ctx, &d, now) before := d.Confidence r.gateLLMDecision(&d) @@ -239,6 +242,7 @@ func (r *Router) Route(ctx context.Context, utterance string, now time.Time) (De Intent: best.Intent, Confidence: best.Score, Slots: r.extractor.Extract(ctx, best.Intent, utterance, now), + Producer: RouteProducerClassifier, } // stage 3 — confidence gate. Below threshold ⇒ clarify, don't guess. diff --git a/internal/router/source.go b/internal/router/source.go index 13e9273..0469f08 100644 --- a/internal/router/source.go +++ b/internal/router/source.go @@ -76,3 +76,26 @@ func ValidSource(s Source) bool { } return false } + +// InputSource — which channel this utterance arrived on. The same provenance +// vocabulary facts use (internal/event). Threaded through the turn because a +// turn can write a fact, and a fact that lies about where it came from is +// worse than no fact: provenance is the first column read when asking why a +// daemon-wide setting is the way it is. +type InputSource string + +const ( + // InputSourceVoice — a real microphone (PushToTalk). + InputSourceVoice InputSource = "tap:voice" + // InputSourceText — mavweb /api/chat, telegram, or any text entry point. + InputSourceText InputSource = "tap:text" +) + +// NormalizedInput — the typed ingress boundary for a turn. Text is the raw +// utterance after STT (voice) or as typed (text). Source identifies the +// channel. This slice performs no new linguistic normalization: text and voice +// paths continue to converge onto the same turn path as they did before. +type NormalizedInput struct { + Text string + Source InputSource +} diff --git a/internal/store/migrations.go b/internal/store/migrations.go index 65c7526..435eaf5 100644 --- a/internal/store/migrations.go +++ b/internal/store/migrations.go @@ -385,6 +385,11 @@ ALTER TABLE reminders ADD COLUMN next_fire_ts INTEGER;`, // #2 CREATE UNIQUE INDEX IF NOT EXISTS idx_digest_entries_live_candidate ON digest_entries (rule, candidate_fingerprint) WHERE status = 'pending' AND candidate_fingerprint <> '';`, + + // #27 — route_producer on routing_traces. Records which cascade stage + // (grammar, heads, llm, classifier) produced the routing decision, so a + // trace can name the winning component directly. + `ALTER TABLE routing_traces ADD COLUMN route_producer TEXT NOT NULL DEFAULT '';`, } // migrate applies every migration with a number greater than the DB's current diff --git a/internal/store/routingtraces.go b/internal/store/routingtraces.go index bdf83a9..06c4730 100644 --- a/internal/store/routingtraces.go +++ b/internal/store/routingtraces.go @@ -30,6 +30,9 @@ type RoutingTrace struct { Source string `json:"source"` Winner string `json:"winner"` Intent string `json:"intent"` + // RouteProducer — which cascade stage produced the routing decision. + // grammar, heads, llm, or classifier. Empty on pre-route turns. + RouteProducer string `json:"route_producer,omitempty"` // ClaimedBeforeHead — stage 0 or a pre-route resolver answered, so the turn // teaches nothing about the classifier. It is a large share of real traffic, // and counting those turns as training signal would fit the head to the @@ -56,10 +59,10 @@ func (s *Store) WriteRoutingTrace(ctx context.Context, tr RoutingTrace) (int64, } res, err := s.db.ExecContext(ctx, ` INSERT INTO routing_traces - (ts, utterance, source, winner, intent, claimed_before_head, encoder_id, outcome, correction, claims) - VALUES (?,?,?,?,?,?,?,?,?,?)`, + (ts, utterance, source, winner, intent, route_producer, claimed_before_head, encoder_id, outcome, correction, claims) + VALUES (?,?,?,?,?,?,?,?,?,?,?)`, tr.Ts.UnixMilli(), tr.Utterance, tr.Source, tr.Winner, tr.Intent, - tr.ClaimedBeforeHead, tr.EncoderID, tr.Outcome, tr.Correction, claims) + tr.RouteProducer, tr.ClaimedBeforeHead, tr.EncoderID, tr.Outcome, tr.Correction, claims) if err != nil { return 0, fmt.Errorf("write routing trace: %w", err) } @@ -91,8 +94,8 @@ func (s *Store) PruneRoutingTraces(ctx context.Context, before time.Time) error // RecentRoutingTraces returns the newest n turns, newest first. func (s *Store) RecentRoutingTraces(ctx context.Context, n int) ([]RoutingTrace, error) { rows, err := s.db.QueryContext(ctx, ` - SELECT id, ts, utterance, source, winner, intent, claimed_before_head, - encoder_id, outcome, correction, claims + SELECT id, ts, utterance, source, winner, intent, route_producer, + claimed_before_head, encoder_id, outcome, correction, claims FROM routing_traces ORDER BY id DESC LIMIT ?`, n) @@ -106,8 +109,8 @@ func (s *Store) RecentRoutingTraces(ctx context.Context, n int) ([]RoutingTrace, var tsMilli int64 var claims string if err := rows.Scan(&tr.ID, &tsMilli, &tr.Utterance, &tr.Source, &tr.Winner, - &tr.Intent, &tr.ClaimedBeforeHead, &tr.EncoderID, &tr.Outcome, - &tr.Correction, &claims); err != nil { + &tr.Intent, &tr.RouteProducer, &tr.ClaimedBeforeHead, &tr.EncoderID, + &tr.Outcome, &tr.Correction, &claims); err != nil { return nil, err } tr.Ts = time.UnixMilli(tsMilli).UTC()