From 4d97280d74fdcf71c4dc9622b7b890dc4c4cf8a6 Mon Sep 17 00:00:00 2001 From: claude Date: Thu, 6 Aug 2026 19:45:37 +0400 Subject: [PATCH] a turn hands back its trace id, and one wire op corrects it (V-630) The correction is the only supervised signal in the box, so the cost of giving one has to be near zero. That means the surface needs the trace id of the turn it is showing, which it had no way to learn: handleText returns one string and the trace was written after the reply left. The id rides back on ChatReply through the same context sink querySource uses, so the mic, telegram and the web keep the one signature they share. CorrectTurn takes a trace id and an optional target, which is deliberately reach-agnostic: nothing about it assumes a browser. store.ErrNoSuchTrace gets a wire twin. A turn past the retention bound is gone, and that is the expected outcome of correcting an old turn, not a broken database. --- cmd/mavend/routingtrace.go | 50 +++++++++++++++++++++++++++++++-- cmd/mavend/routingtrace_test.go | 46 ++++++++++++++++++++++++++++++ cmd/mavend/tick_api.go | 11 +++++++- internal/ipc/api.go | 23 +++++++++++++-- internal/ipc/client.go | 8 +++++- internal/ipc/coreapi.go | 8 ++++++ internal/ipc/maperr_test.go | 1 + internal/ipc/server.go | 5 +++- internal/ipc/storeapi.go | 9 ++++++ internal/ipc/unimplemented.go | 4 +++ internal/ipc/wire.go | 4 +++ 11 files changed, 162 insertions(+), 7 deletions(-) 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):