diff --git a/cmd/mavend/routingtrace.go b/cmd/mavend/routingtrace.go index 02e531b..5c6832e 100644 --- a/cmd/mavend/routingtrace.go +++ b/cmd/mavend/routingtrace.go @@ -16,6 +16,7 @@ import ( "encoding/json" "log" "strings" + "sync" "time" "github.com/kami/maven/internal/decision" @@ -38,6 +39,44 @@ func traceSink(s *store.Store) traceWriter { return s } +// The trace id rides the context, the same seam querysource.go uses and for the +// same reason: handleText answers every reach through one string, and threading +// a second value through the whole action dispatch would change a signature the +// mic, telegram and the web all share. A caller that wants the id asks for a +// sink; the mic path does not, and pays nothing. +type traceIDKey struct{} + +type traceIDSink struct { + mu sync.Mutex + id int64 +} + +func (s *traceIDSink) note(id int64) { + s.mu.Lock() + defer s.mu.Unlock() + s.id = id +} + +// ID is the persisted trace for the turn, or 0 when nothing was persisted. +func (s *traceIDSink) ID() int64 { + s.mu.Lock() + defer s.mu.Unlock() + return s.id +} + +// withTraceIDSink returns a context that collects the persisted trace id, and +// the sink to read after the turn has answered. +func withTraceIDSink(ctx context.Context) (context.Context, *traceIDSink) { + sink := &traceIDSink{} + return context.WithValue(ctx, traceIDKey{}, sink), sink +} + +func noteTraceID(ctx context.Context, id int64) { + if sink, ok := ctx.Value(traceIDKey{}).(*traceIDSink); ok { + sink.note(id) + } +} + // 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 @@ -58,7 +97,8 @@ func pruneTracesOnStart(s *store.Store, now time.Time) { // // 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) { +func (h *reactiveHandler) persistDecision(turnCtx context.Context, rec *decision.Record, src turnSource) { + ctx := turnCtx if h.traces == nil || rec == nil || strings.TrimSpace(rec.Utterance) == "" { return } @@ -85,9 +125,15 @@ func (h *reactiveHandler) persistDecision(ctx context.Context, rec *decision.Rec Outcome: wonAt(rec, decision.StageAction), Claims: claims, } - if _, err := h.traces.WriteRoutingTrace(ctx, tr); err != nil { + id, err := h.traces.WriteRoutingTrace(ctx, tr) + if err != nil { log.Printf("routing trace: write: %v", err) + return } + // The id goes back to whoever asked for it, so /chat can offer a correction + // on the turn it is already showing (V-630). Noted on the ORIGINAL context, + // not the detached one above: the sink belongs to the caller's turn. + noteTraceID(turnCtx, id) } // wonIntent — what the winning claimant made the turn. Read from the claim diff --git a/cmd/mavend/routingtrace_test.go b/cmd/mavend/routingtrace_test.go index 2ed45fb..88aa251 100644 --- a/cmd/mavend/routingtrace_test.go +++ b/cmd/mavend/routingtrace_test.go @@ -77,3 +77,49 @@ func TestEmptyUtteranceIsNotPersisted(t *testing.T) { } var _ traceWriter = (*store.Store)(nil) + +// The trace id rides back to the caller, which is what makes a correction one +// gesture: /chat already has the id, so saying "that was wrong" costs a button +// and no lookup (V-630). +func TestTurnHandsBackItsTraceID(t *testing.T) { + h := traceHandler(t, decision.NewRing()) + h.traces = traceSink(h.dataStore) + + ctx, sink := withTraceIDSink(context.Background()) + if reply := h.handleText(ctx, "web", "сколько сейчас времени"); reply == "" { + t.Fatal("turn produced no reply") + } + id := sink.ID() + if id == 0 { + t.Fatal("no trace id came back, so /chat can offer no correction") + } + // And it names the turn that just ran, so the correction lands on the right + // utterance. + if err := h.dataStore.CorrectTurn(context.Background(), id, "query", h.now()); err != nil { + t.Fatal(err) + } + labels, err := h.dataStore.RoutingLabels(context.Background(), 5) + if err != nil { + t.Fatal(err) + } + if len(labels) != 1 || labels[0].Utterance != "сколько сейчас времени" { + t.Fatalf("labels %+v, want the turn that just ran", labels) + } +} + +// A turn nobody asked the id of costs nothing, which is the mic path. +func TestTurnWithNoSinkStillPersists(t *testing.T) { + h := traceHandler(t, decision.NewRing()) + h.traces = traceSink(h.dataStore) + + 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)) + } +} diff --git a/cmd/mavend/tick_api.go b/cmd/mavend/tick_api.go index 236b521..adb63e4 100644 --- a/cmd/mavend/tick_api.go +++ b/cmd/mavend/tick_api.go @@ -97,10 +97,19 @@ func (d *daemonAPI) Chat(ctx context.Context, conversation, text string) (ipc.Ch return ipc.ChatReply{}, errors.New("mavend: chat not available") } ctx, sink := withQuerySourceSink(ctx) + // The trace id rides back the same way (V-630), so /chat can offer a + // correction on the turn it is already showing. 0 when nothing persisted. + ctx, traces := withTraceIDSink(ctx) reply := d.chatFn(ctx, conversation, text) - return ipc.ChatReply{Reply: reply, Source: sink.Name()}, nil + return ipc.ChatReply{Reply: reply, Source: sink.Name(), TraceID: traces.ID()}, nil } +// CorrectTurn is NOT overridden here, and that is deliberate (V-630). Every other +// diagnostic on this type exists because the daemon holds something the store +// cannot answer from a table. A correction is a table, so the embedded store +// adapter is already the right answer and a second implementation here would be +// a second place for it to drift. + // MCPServers — the configured MCP servers and their health (Vikunja #251). // Empty, not an error, when the mcp block is absent: "not configured" is the // default state and the web surface renders it as such. diff --git a/internal/ipc/api.go b/internal/ipc/api.go index 113c825..db90185 100644 --- a/internal/ipc/api.go +++ b/internal/ipc/api.go @@ -724,8 +724,15 @@ type chatReq struct { Conversation string `json:"conversation,omitempty"` } type chatResp struct { - Reply string `json:"reply"` - Source string `json:"source,omitempty"` + Reply string `json:"reply"` + Source string `json:"source,omitempty"` + TraceID int64 `json:"trace_id,omitempty"` +} + +// correctTurnReq — the owner correcting one persisted turn (V-630). +type correctTurnReq struct { + TraceID int64 `json:"trace_id"` + ShouldBe string `json:"should_be,omitempty"` } // ChatReply — one text turn's answer plus which query source claimed it. @@ -738,6 +745,12 @@ type chatResp struct { type ChatReply struct { Reply string Source string + // TraceID is the persisted routing trace for this turn (V-629), and it is + // what makes a correction one gesture: the surface already has the id, so + // saying "that was wrong" costs a button and no lookup. 0 ⇒ nothing was + // persisted, which is a box with no database, and the surface offers no + // correction rather than a broken one. + TraceID int64 } type proposeToolReq struct { @@ -937,6 +950,12 @@ var ErrTaskDuplicate = errors.New("ipc: another live task already has this text" // is down" must not read the same to a caller deciding whether to store an id. var ErrNoEntity = errors.New("ipc: no such entity") +// ErrNoSuchTrace — the turn a correction names is not in routing_traces. Given +// a wire twin because it is the expected outcome of correcting a turn older than +// the retention bound, and "that turn is gone" and "the database is broken" must +// not read the same to the surface offering the gesture. +var ErrNoSuchTrace = errors.New("ipc: no such routing trace") + // ErrTaskResolved — a resolved task is not editable. var ErrTaskResolved = errors.New("ipc: task is resolved") diff --git a/internal/ipc/client.go b/internal/ipc/client.go index 946ed37..5cf1fc9 100644 --- a/internal/ipc/client.go +++ b/internal/ipc/client.go @@ -247,6 +247,8 @@ func hydrate(e *RpcError) error { return fmt.Errorf("%w: %s", ErrReminderState, e.Message) case codeToolNotFound: return fmt.Errorf("%w: %s", ErrToolNotFound, e.Message) + case codeNoSuchTrace: + return fmt.Errorf("%w: %s", ErrNoSuchTrace, e.Message) case codeUnknownMethod: return fmt.Errorf("%w: %s", ErrUnknownMethod, e.Message) case codeBadParams: @@ -661,7 +663,11 @@ func (c *Client) Chat(ctx context.Context, conversation, text string) (ChatReply if err := c.call(ctx, MethodChat, chatReq{Text: text, Conversation: conversation}, &r); err != nil { return ChatReply{}, err } - return ChatReply{Reply: r.Reply, Source: r.Source}, nil + return ChatReply{Reply: r.Reply, Source: r.Source, TraceID: r.TraceID}, nil +} + +func (c *Client) CorrectTurn(ctx context.Context, traceID int64, shouldBe string) error { + return c.call(ctx, MethodCorrectTurn, correctTurnReq{TraceID: traceID, ShouldBe: shouldBe}, nil) } func (c *Client) TickTrace(ctx context.Context) (TickTrace, error) { diff --git a/internal/ipc/coreapi.go b/internal/ipc/coreapi.go index d3d952d..cf3d5dc 100644 --- a/internal/ipc/coreapi.go +++ b/internal/ipc/coreapi.go @@ -176,6 +176,14 @@ type SystemAPI interface { // has run since the daemon started. TurnDecisions(ctx context.Context, n int) ([]TurnDecision, error) + // CorrectTurn records that one persisted turn was routed wrongly, and what + // it should have been (V-630). shouldBe empty means "wrong, target + // unstated", which is a usable negative and must not cost more to give than + // the full answer. Unlike TurnDecisions this DOES reach a table, because a + // correction is the only supervised signal the box gets and it has to + // outlive the trace that carried it. + CorrectTurn(ctx context.Context, traceID int64, shouldBe string) error + // RecentEcosystemTraces reads the ecosystem call log, which lives in its // own table so machine-rate traces never crowd out human-rate facts. RecentEcosystemTraces(ctx context.Context, n int) ([]EcosystemTrace, error) diff --git a/internal/ipc/maperr_test.go b/internal/ipc/maperr_test.go index 7bdc4b3..6ff31fb 100644 --- a/internal/ipc/maperr_test.go +++ b/internal/ipc/maperr_test.go @@ -31,6 +31,7 @@ var mapErrPairs = []struct { {"ErrReminderNotFound", store.ErrReminderNotFound, ErrReminderNotFound}, {"ErrReminderState", store.ErrReminderState, ErrReminderState}, {"ErrToolNotFound", store.ErrToolNotFound, ErrToolNotFound}, + {"ErrNoSuchTrace", store.ErrNoSuchTrace, ErrNoSuchTrace}, {"ErrTaskNoDoneWhen", store.ErrTaskNoDoneWhen, ErrTaskNoDoneWhen}, {"ErrTaskDuplicate", store.ErrTaskDuplicate, ErrTaskDuplicate}, {"ErrTaskResolved", store.ErrTaskResolved, ErrTaskResolved}, diff --git a/internal/ipc/server.go b/internal/ipc/server.go index 4d035a3..97ee006 100644 --- a/internal/ipc/server.go +++ b/internal/ipc/server.go @@ -514,7 +514,10 @@ var methodTable = map[Method]handlerFunc{ }), MethodChat: withParams(func(ctx context.Context, api CoreAPI, p chatReq) (chatResp, error) { reply, err := api.Chat(ctx, p.Conversation, p.Text) - return chatResp{Reply: reply.Reply, Source: reply.Source}, err + return chatResp{Reply: reply.Reply, Source: reply.Source, TraceID: reply.TraceID}, err + }), + MethodCorrectTurn: withParams(func(ctx context.Context, api CoreAPI, p correctTurnReq) (struct{}, error) { + return struct{}{}, api.CorrectTurn(ctx, p.TraceID, p.ShouldBe) }), MethodTickTrace: withoutParams(func(ctx context.Context, api CoreAPI) (TickTrace, error) { return api.TickTrace(ctx) diff --git a/internal/ipc/storeapi.go b/internal/ipc/storeapi.go index 21fd7da..a0fe3f2 100644 --- a/internal/ipc/storeapi.go +++ b/internal/ipc/storeapi.go @@ -213,6 +213,13 @@ func (a *storeAPI) TickTrace(ctx context.Context) (TickTrace, error) { return TickTrace{}, errors.New("store: tick trace not available via direct store API") } +// CorrectTurn — unlike TickTrace and TurnDecisions this one is a table, so the +// store adapter answers it for real (V-630). A correction has to land whether +// the caller reached the daemon or the store directly. +func (a *storeAPI) CorrectTurn(ctx context.Context, traceID int64, shouldBe string) error { + return mapErr(a.s.CorrectTurn(ctx, traceID, shouldBe, time.Now())) +} + // TurnDecisions — same story as TickTrace: the arbitration record is a daemon // ring, not a table, so there is nothing here to read it from (V-564). func (a *storeAPI) TurnDecisions(ctx context.Context, n int) ([]TurnDecision, error) { @@ -412,6 +419,8 @@ func mapErr(err error) error { return ErrReminderState case errors.Is(err, store.ErrToolNotFound): return ErrToolNotFound + case errors.Is(err, store.ErrNoSuchTrace): + return ErrNoSuchTrace case errors.Is(err, store.ErrTaskNoDoneWhen): return ErrTaskNoDoneWhen case errors.Is(err, store.ErrTaskDuplicate): diff --git a/internal/ipc/unimplemented.go b/internal/ipc/unimplemented.go index 69681cb..e5837bc 100644 --- a/internal/ipc/unimplemented.go +++ b/internal/ipc/unimplemented.go @@ -144,6 +144,10 @@ func (UnimplementedCoreAPI) RevertFact(ctx context.Context, key string) (int64, func (UnimplementedCoreAPI) TickTrace(ctx context.Context) (TickTrace, error) { return TickTrace{}, ErrNotImplemented } +func (UnimplementedCoreAPI) CorrectTurn(ctx context.Context, traceID int64, shouldBe string) error { + return ErrNotImplemented +} + func (UnimplementedCoreAPI) TurnDecisions(ctx context.Context, n int) ([]TurnDecision, error) { return nil, ErrNotImplemented } diff --git a/internal/ipc/wire.go b/internal/ipc/wire.go index 4b6ac78..a2058b0 100644 --- a/internal/ipc/wire.go +++ b/internal/ipc/wire.go @@ -49,6 +49,7 @@ const ( MethodRevertFact Method = "revert_fact" MethodTickTrace Method = "tick_trace" MethodTurnDecisions Method = "turn_decisions" + MethodCorrectTurn Method = "correct_turn" MethodMorningStatus Method = "morning_status" MethodMCPServers Method = "mcp_servers" MethodDayPlan Method = "day_plan" @@ -125,6 +126,7 @@ const ( codeReminderMissing = "reminder_not_found" codeReminderState = "reminder_state" codeToolNotFound = "tool_not_found" + codeNoSuchTrace = "no_such_trace" codeUnknownMethod = "unknown_method" codeBadParams = "bad_params" codeForbidden = "forbidden" @@ -160,6 +162,8 @@ func codeOf(err error) string { return codeReminderState case errors.Is(err, ErrToolNotFound): return codeToolNotFound + case errors.Is(err, ErrNoSuchTrace): + return codeNoSuchTrace case errors.Is(err, ErrUnknownMethod): return codeUnknownMethod case errors.Is(err, ErrBadParams):