Merge the persisted routing trace (#183)
This commit was merged in pull request #183.
This commit is contained in:
@@ -285,8 +285,22 @@ site cannot change a route and a context with no record costs nothing. It is
|
||||
installed in `runTurn`, so the mic, telegram and the web all leave the same
|
||||
trail. Storage is a 25-turn in-memory ring on the handler (`decision.Ring`),
|
||||
read over `ipc.TurnDecisions` and rendered as the second table on `/trace`.
|
||||
Nothing persists: a turn record is read minutes later or never, and his words do
|
||||
not belong in a table that outlives the diagnosis. Adding a rung to the ladder
|
||||
**It also persists, since 06-08-2026, and that reverses what this section used to
|
||||
say** (V-629, `docs/plans/21-persisting-the-routing-trace.md`). The old rule was
|
||||
that nothing persists, because a turn record is read minutes later or never. The
|
||||
owner reversed it: the routing heads (V-546) cannot be fitted or calibrated
|
||||
without real utterances, and 9 of the 31 modes in `internal/modes` have no seed
|
||||
example at all. The ring did not move. It is still what `/trace` reads and still
|
||||
what a test with no store gets. `cmd/mavend/routingtrace.go` is a second sink
|
||||
beside it, writing `routing_traces` (migration #23). The utterance is stored in
|
||||
clear, because a 384-dimension vector of a short sentence is substantially
|
||||
recoverable and storing vectors instead would be a privacy claim we cannot
|
||||
support. What makes it safe is the same thing that makes the fact store safe.
|
||||
Retention is 14 days, enforced on write. Nothing reads it outward, and the rule
|
||||
that his notes and facts are never search input covers this table. `Store.Wipe`
|
||||
deletes it with everything else. A correction (V-630) is promoted out into a
|
||||
seed-shaped row and kept, because a label is not a transcript. The transcript
|
||||
still expires. Adding a rung to the ladder
|
||||
in `runTurn` means adding its name to `preRouteLadder` in
|
||||
`cmd/mavend/decisiontrace.go`, or that rung is silently missing from the record.
|
||||
|
||||
|
||||
@@ -0,0 +1,128 @@
|
||||
// mavend/routingtrace.go — persisting the per-turn decision record (V-629).
|
||||
//
|
||||
// internal/decision keeps a 25-turn in-memory ring and persisted nothing, on the
|
||||
// argument that a turn record is read minutes later or never. The owner reversed
|
||||
// that on 06-08-2026, because the routing heads (V-546) cannot be fitted or
|
||||
// calibrated without real utterances and there is no other source of them. The
|
||||
// reversal is written down in docs/plans/21-persisting-the-routing-trace.md.
|
||||
//
|
||||
// The ring stays. It is what /trace reads, it is fast, and it is what a test that
|
||||
// wired no store still gets. This file is the second sink beside it, and it is
|
||||
// nil unless the daemon has a database — no store, no trace, no error.
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"log"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/kami/maven/internal/decision"
|
||||
"github.com/kami/maven/internal/store"
|
||||
)
|
||||
|
||||
// traceWriter is the seam the handler persists through. store.Store satisfies
|
||||
// it. nil ⇒ the ring is the only sink, which is the pre-V-629 behaviour exactly.
|
||||
type traceWriter interface {
|
||||
WriteRoutingTrace(ctx context.Context, tr store.RoutingTrace) (int64, error)
|
||||
}
|
||||
|
||||
// traceSink wraps the store, or returns nil when there is none. A typed nil
|
||||
// pointer assigned straight into the interface would be non-nil and would panic
|
||||
// on the first turn, which is the classic shape of this bug.
|
||||
func traceSink(s *store.Store) traceWriter {
|
||||
if s == nil {
|
||||
return nil
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// pruneTracesOnStart enforces retention once at wiring time. Pruning on write
|
||||
// alone is not enough: a box that goes quiet for a month keeps every row until
|
||||
// the next sixty-fourth turn, and "kept for fourteen days" would then be true
|
||||
// only of a box in daily use. Called for its effect and never blocks a start.
|
||||
func pruneTracesOnStart(s *store.Store, now time.Time) {
|
||||
if s == nil {
|
||||
return
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
if err := s.PruneRoutingTraces(ctx, now.Add(-store.RoutingTraceRetention)); err != nil {
|
||||
log.Printf("routing trace: prune on start: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// persistDecision writes one finished record. It takes the same *decision.Record
|
||||
// the ring takes, so the two sinks cannot disagree about what the turn did.
|
||||
//
|
||||
// Errors are logged and swallowed. A trace is diagnostic and training data, and
|
||||
// a failed insert must never change what the owner hears.
|
||||
func (h *reactiveHandler) persistDecision(ctx context.Context, rec *decision.Record, src turnSource) {
|
||||
if h.traces == nil || rec == nil || strings.TrimSpace(rec.Utterance) == "" {
|
||||
return
|
||||
}
|
||||
// Detached from the turn's context, and bounded on its own. Two reasons, and
|
||||
// the first is the one that matters: the turn is over by the time this runs,
|
||||
// so a caller that hung up or timed out would cancel the insert, and the turn
|
||||
// he abandoned halfway is exactly the one worth having. The second is that a
|
||||
// write must not hold the reply, so it gets a second and no more.
|
||||
ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), time.Second)
|
||||
defer cancel()
|
||||
claims, err := json.Marshal(rec.Claims)
|
||||
if err != nil {
|
||||
log.Printf("routing trace: marshal claims: %v", err)
|
||||
return
|
||||
}
|
||||
tr := store.RoutingTrace{
|
||||
Ts: rec.Ts,
|
||||
Utterance: rec.Utterance,
|
||||
Source: string(src),
|
||||
Winner: rec.Winner,
|
||||
Intent: wonIntent(rec),
|
||||
ClaimedBeforeHead: claimedBeforeHead(rec),
|
||||
EncoderID: h.encoderID,
|
||||
Outcome: wonAt(rec, decision.StageAction),
|
||||
Claims: claims,
|
||||
}
|
||||
if _, err := h.traces.WriteRoutingTrace(ctx, tr); err != nil {
|
||||
log.Printf("routing trace: write: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// wonIntent — what the winning claimant made the turn. Read from the claim
|
||||
// rather than from the route, because a pre-route resolver wins without routing
|
||||
// and its intent is the honest answer to "what was this turn".
|
||||
func wonIntent(rec *decision.Record) string {
|
||||
for _, c := range rec.Claims {
|
||||
if c.Outcome == decision.Won && c.Intent != "" {
|
||||
return c.Intent
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// wonAt — the claimant that won at one stage. The action stage is what actually
|
||||
// produced the reply, which is a different question from what was routed: a
|
||||
// route that reached a gap and a route that ran are not the same turn.
|
||||
func wonAt(rec *decision.Record, stage string) string {
|
||||
for _, c := range rec.Claims {
|
||||
if c.Stage == stage && c.Outcome == decision.Won {
|
||||
return c.Claimant
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// claimedBeforeHead — a pre-route resolver or a stage-0 grammar answered, so the
|
||||
// turn teaches nothing about the classifier. Those are a large share of real
|
||||
// traffic, and fitting a head on them would fit it to the grammars rather than
|
||||
// to him. Recorded per turn rather than filtered on write, because which share
|
||||
// that is happens to be the number V-632 needs to know.
|
||||
func claimedBeforeHead(rec *decision.Record) bool {
|
||||
stage, _, ok := strings.Cut(rec.Winner, ":")
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
return stage == decision.StagePreRoute || stage == decision.StageZero
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/kami/maven/internal/decision"
|
||||
"github.com/kami/maven/internal/store"
|
||||
)
|
||||
|
||||
// A real turn leaves a persisted trace, not only a ring entry. This is the whole
|
||||
// of V-629: without one there is nothing to fit the routing heads from.
|
||||
func TestTurnPersistsTrace(t *testing.T) {
|
||||
ring := decision.NewRing()
|
||||
h := traceHandler(t, ring)
|
||||
h.traces = traceSink(h.dataStore)
|
||||
h.encoderID = "hash-1024"
|
||||
|
||||
if reply := h.handleText(context.Background(), "web", "сколько сейчас времени"); reply == "" {
|
||||
t.Fatal("turn produced no reply")
|
||||
}
|
||||
got, err := h.dataStore.RecentRoutingTraces(context.Background(), 5)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(got) != 1 {
|
||||
t.Fatalf("persisted %d traces, want 1", len(got))
|
||||
}
|
||||
tr := got[0]
|
||||
if tr.Utterance != "сколько сейчас времени" {
|
||||
t.Errorf("utterance %q", tr.Utterance)
|
||||
}
|
||||
if tr.Source != string(sourceText) {
|
||||
t.Errorf("source %q, want %q", tr.Source, sourceText)
|
||||
}
|
||||
// A stage-0 clock rule answers this one, so the turn teaches the classifier
|
||||
// nothing and the trace has to say so.
|
||||
if !tr.ClaimedBeforeHead {
|
||||
t.Errorf("claimed_before_head false on winner %q", tr.Winner)
|
||||
}
|
||||
if tr.EncoderID != "hash-1024" {
|
||||
t.Errorf("encoder_id %q", tr.EncoderID)
|
||||
}
|
||||
if len(tr.Claims) < 3 {
|
||||
t.Errorf("claims %s: the losers and the never-asked are the point", tr.Claims)
|
||||
}
|
||||
}
|
||||
|
||||
// No store, no trace, and no panic. A typed nil pointer in the interface would
|
||||
// pass the nil check and die on the first turn.
|
||||
func TestNoStoreNoTrace(t *testing.T) {
|
||||
ring := decision.NewRing()
|
||||
h := traceHandler(t, ring)
|
||||
h.traces = traceSink(nil)
|
||||
|
||||
if reply := h.handleText(context.Background(), "web", "сколько сейчас времени"); reply == "" {
|
||||
t.Fatal("turn produced no reply")
|
||||
}
|
||||
if len(ring.Recent(5)) != 1 {
|
||||
t.Error("the ring is still the first sink and must still hold the turn")
|
||||
}
|
||||
}
|
||||
|
||||
// An empty utterance writes nothing. A blank row carries no label and no
|
||||
// diagnosis, and it is his words the retention bound exists for.
|
||||
func TestEmptyUtteranceIsNotPersisted(t *testing.T) {
|
||||
h := traceHandler(t, decision.NewRing())
|
||||
h.traces = traceSink(h.dataStore)
|
||||
h.persistDecision(context.Background(), &decision.Record{Utterance: " "}, sourceText)
|
||||
got, err := h.dataStore.RecentRoutingTraces(context.Background(), 5)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(got) != 0 {
|
||||
t.Fatalf("persisted %d traces for a blank utterance", len(got))
|
||||
}
|
||||
}
|
||||
|
||||
var _ traceWriter = (*store.Store)(nil)
|
||||
+16
-1
@@ -145,6 +145,17 @@ type reactiveHandler struct {
|
||||
// is recorded, which is what a test that did not ask for one gets.
|
||||
decisions *decision.Ring
|
||||
|
||||
// traces persists those same records (V-629, routingtrace.go). The ring is
|
||||
// still what /trace reads; this is the second sink, and it exists because the
|
||||
// routing heads cannot be fitted without real utterances. nil ⇒ the ring
|
||||
// alone, which is the behaviour every box had before 06-08-2026.
|
||||
traces traceWriter
|
||||
|
||||
// encoderID names the encoder body live on this box, stored beside each
|
||||
// trace: a fitted distance means nothing under another body. Empty ⇒ no
|
||||
// embedder, so the classifier was the keyword floor.
|
||||
encoderID string
|
||||
|
||||
// clarifyStore parks the request behind an open question she asked (see
|
||||
// clarify.go). nil ⇒ she falls back to the canned "не поняла" reply.
|
||||
clarifyStore *dialogue.ClarifyStore
|
||||
@@ -265,7 +276,11 @@ func (h *reactiveHandler) runTurn(ctx context.Context, text string, src turnSour
|
||||
var rec *decision.Record
|
||||
ctx, rec = decision.With(ctx, text)
|
||||
decision.Expect(ctx, decision.StagePreRoute, preRouteLadder)
|
||||
defer func() { h.decisions.Push(rec.Finish(h.now())) }()
|
||||
defer func() {
|
||||
done := rec.Finish(h.now())
|
||||
h.decisions.Push(done)
|
||||
h.persistDecision(ctx, done, src)
|
||||
}()
|
||||
}
|
||||
|
||||
// 0b. the turn's routing, computed at most once and shared (Vikunja #560).
|
||||
|
||||
+10
-1
@@ -149,6 +149,9 @@ func wireVoice(cfg *config.Config, coreAPI ipc.CoreAPI, phr phraser.Phraser, mem
|
||||
w.embedder = emb
|
||||
repairFactVectors(dataStore, emb)
|
||||
checkStoredEmbedder(dataStore, emb)
|
||||
// Retention is enforced on write, which is not enough on its own: a box that
|
||||
// goes quiet keeps every trace until the next sixty-fourth turn (V-629).
|
||||
pruneTracesOnStart(dataStore, time.Now())
|
||||
|
||||
// ----- tool executor (the enabled act allowlist, store-backed) -----
|
||||
// Config tools are the declarative bootstrap: seed them into the store as
|
||||
@@ -298,7 +301,13 @@ func wireVoice(cfg *config.Config, coreAPI ipc.CoreAPI, phr phraser.Phraser, mem
|
||||
// Always on (V-564). The record is the instrument the rest of V-558 is
|
||||
// measured with, and one that only runs when a flag is set is not there
|
||||
// on the night the misroute happens.
|
||||
decisions: decision.NewRing(),
|
||||
decisions: decision.NewRing(),
|
||||
// The second sink (V-629). Same records, persisted, because the routing
|
||||
// heads cannot be fitted from a 25-turn ring. Nil store ⇒ ring only, and
|
||||
// EmbedderID is the same string the vector marker uses, so a trace and a
|
||||
// stored vector name their body the same way.
|
||||
traces: traceSink(dataStore),
|
||||
encoderID: router.EmbedderID(emb),
|
||||
clarifyStore: clarifyStore,
|
||||
// 0 here (unset config) ⇒ the dialogue default.
|
||||
clarifyMaxAttempts: cfg.Voice.ClarifyMaxAttempts,
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
# Plan: persist the routing trace
|
||||
|
||||
**Owner's call, 06-08-2026. Vikunja #629, umbrella #628.**
|
||||
|
||||
**Verdict: the per-turn decision record now persists.** That reverses a written decision,
|
||||
which is the point of this file. It is not an incidental telemetry
|
||||
feature. Do not read it as one.
|
||||
|
||||
Last verified: 06-08-2026 @ 799cf55
|
||||
|
||||
## What the old decision said
|
||||
|
||||
`internal/decision` kept a 25-turn in-memory ring and persisted nothing. The argument was
|
||||
in `CLAUDE.md` and it was a good one. A turn record is read minutes after the turn or
|
||||
never, so a table that outlives the diagnosis buys nothing. His words did not belong in it.
|
||||
|
||||
## Why it reversed
|
||||
|
||||
V-546 replaces the generative router with classification heads on e5-small. Fitting
|
||||
prototypes and calibrating a distance both need real utterances. V-631 measured how few
|
||||
there are. Nine of the 31 modes in `internal/modes` have no seed example at all, and they
|
||||
are exactly the nine with no deterministic matcher. The seed corpus cannot supply them. A
|
||||
seed row is a phrase someone wrote for a matcher, not a thing he said. The 202 generated
|
||||
contrast pairs were tried and cost four points of fixture accuracy.
|
||||
|
||||
So the choice was between no routing heads and a persisted trace. The owner chose the trace.
|
||||
|
||||
## Retention, and why it is two answers
|
||||
|
||||
**Raw trace: 14 days.** `store.RoutingTraceRetention` in `internal/store/routingtraces.go`. That
|
||||
is the life of a diagnosis with room for a weekend. The bound is an age and not a row
|
||||
count. The useful question is what she did this week, and a busy Tuesday must not push last
|
||||
Friday out.
|
||||
|
||||
**A correction: indefinite.** The owner corrects a turn on `/chat` (V-630). The pair is then
|
||||
promoted out of the trace into a seed-shaped row and kept, because a label is not a
|
||||
transcript. What stays in `routing_traces` is the transcript. It expires on the same 14
|
||||
days as every other row, corrected or not.
|
||||
|
||||
## What keeps it safe
|
||||
|
||||
The utterance is stored in clear. A 384-dimension vector of a short sentence is
|
||||
substantially recoverable. Storing vectors instead would be a privacy claim we cannot
|
||||
support, and making it would be worse than staying silent.
|
||||
|
||||
- **Nothing here leaves the box.** The rule that the owner's notes and facts are never
|
||||
search input covers this table too. No query source reads it, and no upstream engine can.
|
||||
- **Retention is enforced on write and again at start.** `WriteRoutingTrace` prunes every
|
||||
64th row, which is hours at human rate. `pruneTracesOnStart` covers the case write alone
|
||||
cannot. A box that goes quiet keeps every row until the next sixty-fourth turn. Without
|
||||
the start-time prune, the bound would hold only for a box in daily use.
|
||||
- **Deletion already exists.** `Store.Wipe` drops every table the database reports, so
|
||||
`mavend -wipe -confirm-wipe` covers this one with no list to edit.
|
||||
- **The ring did not move.** It is still what `/trace` reads and still what a test with no
|
||||
store gets. The table is a second sink beside it. A failed insert is logged and swallowed,
|
||||
because a trace must never change what he hears.
|
||||
|
||||
## What is not decided
|
||||
|
||||
Whether some utterances must never be promoted into a durable label, no matter how badly
|
||||
they routed. That is a content rule and it belongs beside the personal boundary, not in the trace
|
||||
writer. Recorded here, left to the owner.
|
||||
@@ -300,6 +300,36 @@ ALTER TABLE reminders ADD COLUMN next_fire_ts INTEGER;`, // #2
|
||||
// here and no caller has to tell them apart.
|
||||
`ALTER TABLE tasks ADD COLUMN done_when TEXT NOT NULL DEFAULT '';
|
||||
ALTER TABLE tasks ADD COLUMN blocked_on TEXT NOT NULL DEFAULT '';`,
|
||||
// #23 — the routing trace (V-629). internal/decision kept a 25-turn ring and
|
||||
// persisted nothing, on the argument that a turn record is read minutes later
|
||||
// or never. The owner reversed that on 06-08-2026: mode discovery and distance
|
||||
// calibration need real utterances, and there is no other source of them.
|
||||
// docs/plans/21-persisting-the-routing-trace.md carries the
|
||||
// reversal.
|
||||
//
|
||||
// utterance holds his words in clear. A 384-dimension vector of a short
|
||||
// sentence is substantially recoverable, so storing vectors instead would be a
|
||||
// privacy claim we cannot support. What makes it safe is the same thing that
|
||||
// makes the fact store safe: it never leaves the box, retention is bounded at
|
||||
// store.RoutingTraceRetention, and Wipe drops it with everything else.
|
||||
//
|
||||
// correction is empty until the owner corrects a turn on /chat (V-630). A
|
||||
// corrected pair is promoted out of here into a seed-shaped row and kept, so
|
||||
// this column is a queue, not the durable label.
|
||||
`CREATE TABLE IF NOT EXISTS routing_traces (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
ts INTEGER NOT NULL,
|
||||
utterance TEXT NOT NULL,
|
||||
source TEXT NOT NULL DEFAULT '',
|
||||
winner TEXT NOT NULL DEFAULT '',
|
||||
intent TEXT NOT NULL DEFAULT '',
|
||||
claimed_before_head INTEGER NOT NULL DEFAULT 0,
|
||||
encoder_id TEXT NOT NULL DEFAULT '',
|
||||
outcome TEXT NOT NULL DEFAULT '',
|
||||
correction TEXT NOT NULL DEFAULT '',
|
||||
claims TEXT NOT NULL DEFAULT '[]'
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_routing_traces_ts ON routing_traces (ts DESC);`,
|
||||
}
|
||||
|
||||
// migrate applies every migration with a number greater than the DB's current
|
||||
|
||||
@@ -0,0 +1,118 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
// RoutingTraceRetention is how long a raw trace lives (owner's call,
|
||||
// 06-08-2026). A trace is read within a day or two of the turn that produced it,
|
||||
// or never, so two weeks is diagnosis with room for a weekend. It is deliberately
|
||||
// an age and not a row count: the useful question is "what did she do this week",
|
||||
// and a busy Tuesday must not push last Friday out.
|
||||
//
|
||||
// A correction is not covered by this bound. The moment the owner corrects a
|
||||
// turn, the pair is promoted out of the trace into a seed-shaped row and kept
|
||||
// indefinitely, because a label is not a transcript. Keeping the transcript that
|
||||
// carried it would defeat the point of the bound.
|
||||
const RoutingTraceRetention = 14 * 24 * time.Hour
|
||||
|
||||
// RoutingTrace is one turn's arbitration, persisted. It is internal/decision's
|
||||
// Record plus the four things the ring never had to carry: which reach the
|
||||
// utterance arrived on, whether stage 0 answered before the classifier was
|
||||
// consulted, which encoder body was live, and what the turn actually did.
|
||||
type RoutingTrace struct {
|
||||
ID int64 `json:"id"`
|
||||
Ts time.Time `json:"ts"`
|
||||
Utterance string `json:"utterance"`
|
||||
Source string `json:"source"`
|
||||
Winner string `json:"winner"`
|
||||
Intent string `json:"intent"`
|
||||
// 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
|
||||
// grammars rather than to him.
|
||||
ClaimedBeforeHead bool `json:"claimed_before_head"`
|
||||
// EncoderID names the encoder body that was live. A fitted distance means
|
||||
// nothing under another body, and V-546 trains a copy of the weights.
|
||||
EncoderID string `json:"encoder_id"`
|
||||
// Outcome is what happened, not what was routed: a route that reached a gap
|
||||
// and a route that ran are different turns.
|
||||
Outcome string `json:"outcome"`
|
||||
// Correction is the owner's label, empty until he gives one (V-630).
|
||||
Correction string `json:"correction"`
|
||||
// Claims is internal/decision's per-claimant detail, stored as JSON because
|
||||
// nothing queries inside it: it is read whole, beside the turn it explains.
|
||||
Claims json.RawMessage `json:"claims"`
|
||||
}
|
||||
|
||||
// WriteRoutingTrace appends one turn and drops the ones past retention.
|
||||
func (s *Store) WriteRoutingTrace(ctx context.Context, tr RoutingTrace) (int64, error) {
|
||||
claims := "[]"
|
||||
if len(tr.Claims) > 0 {
|
||||
claims = string(tr.Claims)
|
||||
}
|
||||
res, err := s.db.ExecContext(ctx, `
|
||||
INSERT INTO routing_traces
|
||||
(ts, utterance, source, winner, intent, 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)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("write routing trace: %w", err)
|
||||
}
|
||||
id, err := res.LastInsertId()
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("last insert id: %w", err)
|
||||
}
|
||||
// Prune rarely. Turns arrive at human rate, so the bound is a ceiling and
|
||||
// paying for a delete on every one of them buys nothing. 64 turns is hours.
|
||||
if id%64 == 0 {
|
||||
if err := s.PruneRoutingTraces(ctx, tr.Ts.Add(-RoutingTraceRetention)); err != nil {
|
||||
return id, err
|
||||
}
|
||||
}
|
||||
return id, nil
|
||||
}
|
||||
|
||||
// PruneRoutingTraces deletes every trace older than before. A corrected turn is
|
||||
// deleted with the rest: the label was promoted out when the owner wrote it, so
|
||||
// what is left here is the transcript, and the transcript is what expires.
|
||||
func (s *Store) PruneRoutingTraces(ctx context.Context, before time.Time) error {
|
||||
if _, err := s.db.ExecContext(ctx,
|
||||
`DELETE FROM routing_traces WHERE ts < ?`, before.UnixMilli()); err != nil {
|
||||
return fmt.Errorf("prune routing traces: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// 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
|
||||
FROM routing_traces
|
||||
ORDER BY id DESC
|
||||
LIMIT ?`, n)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("recent routing traces: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []RoutingTrace
|
||||
for rows.Next() {
|
||||
var tr 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 {
|
||||
return nil, err
|
||||
}
|
||||
tr.Ts = time.UnixMilli(tsMilli).UTC()
|
||||
tr.Claims = json.RawMessage(claims)
|
||||
out = append(out, tr)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
@@ -0,0 +1,75 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestRoutingTraceRoundTrip(t *testing.T) {
|
||||
s := newTestStore(t)
|
||||
ctx := context.Background()
|
||||
now := time.Date(2026, 8, 6, 12, 0, 0, 0, time.UTC)
|
||||
|
||||
in := RoutingTrace{
|
||||
Ts: now,
|
||||
Utterance: "напомни в 11:00 позвонить маме",
|
||||
Source: "tap:voice",
|
||||
Winner: "stage0:reminder-grammar",
|
||||
Intent: "reminder",
|
||||
ClaimedBeforeHead: true,
|
||||
EncoderID: "e5-small",
|
||||
Outcome: "reminder",
|
||||
Claims: []byte(`[{"stage":"stage0","claimant":"reminder-grammar","outcome":"won"}]`),
|
||||
}
|
||||
if _, err := s.WriteRoutingTrace(ctx, in); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got, err := s.RecentRoutingTraces(ctx, 10)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(got) != 1 {
|
||||
t.Fatalf("got %d traces, want 1", len(got))
|
||||
}
|
||||
// The utterance is stored in clear on purpose: a vector is not redaction.
|
||||
if got[0].Utterance != in.Utterance {
|
||||
t.Errorf("utterance %q, want %q", got[0].Utterance, in.Utterance)
|
||||
}
|
||||
if !got[0].ClaimedBeforeHead {
|
||||
t.Error("claimed_before_head lost, and V-632 needs exactly that share")
|
||||
}
|
||||
if got[0].EncoderID != in.EncoderID {
|
||||
t.Errorf("encoder_id %q, want %q", got[0].EncoderID, in.EncoderID)
|
||||
}
|
||||
if string(got[0].Claims) != string(in.Claims) {
|
||||
t.Errorf("claims %s, want %s", got[0].Claims, in.Claims)
|
||||
}
|
||||
if got[0].Correction != "" {
|
||||
t.Errorf("correction %q on an uncorrected turn", got[0].Correction)
|
||||
}
|
||||
}
|
||||
|
||||
// The bound is an age, not a row count: the useful question is what she did this
|
||||
// week, and a busy Tuesday must not push last Friday out.
|
||||
func TestPruneRoutingTracesByAge(t *testing.T) {
|
||||
s := newTestStore(t)
|
||||
ctx := context.Background()
|
||||
now := time.Date(2026, 8, 6, 12, 0, 0, 0, time.UTC)
|
||||
|
||||
for _, age := range []time.Duration{0, 13 * 24 * time.Hour, 15 * 24 * time.Hour} {
|
||||
if _, err := s.WriteRoutingTrace(ctx, RoutingTrace{Ts: now.Add(-age), Utterance: "привет"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err := s.PruneRoutingTraces(ctx, now.Add(-RoutingTraceRetention)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got, err := s.RecentRoutingTraces(ctx, 10)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(got) != 2 {
|
||||
t.Fatalf("kept %d traces, want the two inside 14 days", len(got))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user