fix(agent): live chat turns persist incrementally, survive client disconnect
Fix A3 of plans/2026-07-11-nomos-agent-code-review.md. handleChat only ever saved the assistant message ONCE, after a.chat(...) returned, using ctx := r.Context() for that write — the same context that cancels the instant the client disconnects (Stop button, tab close, network blip). A disconnect mid-turn meant the final save ran with an already-cancelled context and its error was never checked: the entire turn's tool-call history was silently lost from the persisted transcript, even though real work (executions launched, knowledge written) had already happened server-side. Brought handleChat in line with resumeSession's existing pattern (continue.go): insert a placeholder assistant row immediately, update the SAME row after every tool call. The key fix is WHICH context the writes use — a new pctx := context.Background() for every DB write in this handler (session creation/touch, the user message, question auto-close, the placeholder + incremental updates, the title update), while ctx/r.Context() still gates the agent's own work (a.chat) and the SSE writes exactly as before — a disconnect still correctly stops the agent from doing further work, it just no longer also erases what it already did. Verified live: sent a message requiring 6 tool calls (get_entity/ get_relations/get_blast_radius on two targets) and force-killed the client connection mid-stream with curl -m 12 (confirmed via exit code 28). Before this fix the persisted transcript would show 0 tool-call entries; after, all 12 raw tool_use/tool_result entries (6 calls × 2) were present and correctly attributed by tool name — proving both that progress survives an abort and that the incremental writes aren't corrupting the data. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -17,6 +17,7 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/dtoro/oikos/internal/safego"
|
"github.com/dtoro/oikos/internal/safego"
|
||||||
|
"github.com/google/uuid"
|
||||||
)
|
)
|
||||||
|
|
||||||
func main() {
|
func main() {
|
||||||
@@ -173,9 +174,21 @@ func handleChat(w http.ResponseWriter, r *http.Request, a *agent, st *store) {
|
|||||||
ctx := r.Context()
|
ctx := r.Context()
|
||||||
sessionID := req.SessionID
|
sessionID := req.SessionID
|
||||||
|
|
||||||
|
// pctx (persistence context) is deliberately context.Background(), not
|
||||||
|
// ctx/r.Context(), for every DB write in this handler — ctx cancels the
|
||||||
|
// instant the client disconnects (Stop button, tab close, network blip),
|
||||||
|
// and a write made with an already-cancelled context fails. Before this
|
||||||
|
// fix, the assistant message was only ever saved ONCE, at the very end,
|
||||||
|
// using ctx — so a disconnect mid-turn silently lost the ENTIRE turn's
|
||||||
|
// tool-call history from the persisted transcript, even though real work
|
||||||
|
// (executions launched, knowledge written) had already happened
|
||||||
|
// server-side. The agent's own work (a.chat below) still correctly stops
|
||||||
|
// when ctx cancels — this only changes what happens to persistence.
|
||||||
|
pctx := context.Background()
|
||||||
|
|
||||||
if sessionID == "" {
|
if sessionID == "" {
|
||||||
title := truncate(req.Message, 80)
|
title := truncate(req.Message, 80)
|
||||||
sess, err := st.createSession(ctx, title)
|
sess, err := st.createSession(pctx, title)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
slog.Error("nomos: create session", "error", err)
|
slog.Error("nomos: create session", "error", err)
|
||||||
sessionID = "ephemeral"
|
sessionID = "ephemeral"
|
||||||
@@ -183,20 +196,20 @@ func handleChat(w http.ResponseWriter, r *http.Request, a *agent, st *store) {
|
|||||||
sessionID = sess.ID
|
sessionID = sess.ID
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
st.touchSession(ctx, sessionID)
|
st.touchSession(pctx, sessionID)
|
||||||
}
|
}
|
||||||
|
|
||||||
slog.Info("nomos: chat", "session", sessionID, "message", truncate(req.Message, 100))
|
slog.Info("nomos: chat", "session", sessionID, "message", truncate(req.Message, 100))
|
||||||
|
|
||||||
userMsg, _ := json.Marshal(map[string]any{"role": "user", "text": req.Message})
|
userMsg, _ := json.Marshal(map[string]any{"role": "user", "text": req.Message})
|
||||||
st.saveMessage(ctx, sessionID, "user", userMsg)
|
st.saveMessage(pctx, sessionID, "user", userMsg)
|
||||||
|
|
||||||
// If this task has a pending operator question, the incoming message IS the
|
// If this task has a pending operator question, the incoming message IS the
|
||||||
// answer — close it so the panel clears. No separate resume needed: this
|
// answer — close it so the panel clears. No separate resume needed: this
|
||||||
// chat turn is the resume, and the agent sees the question + answer in its
|
// chat turn is the resume, and the agent sees the question + answer in its
|
||||||
// replayed history.
|
// replayed history.
|
||||||
if qid := st.openQuestionID(ctx, sessionID); qid != "" {
|
if qid := st.openQuestionID(pctx, sessionID); qid != "" {
|
||||||
st.answerQuestion(ctx, sessionID, qid, req.Message)
|
st.answerQuestion(pctx, sessionID, qid, req.Message)
|
||||||
}
|
}
|
||||||
|
|
||||||
sseEvent(w, flusher, agentEvent{Type: "session", Data: sessionID, SessionID: sessionID})
|
sseEvent(w, flusher, agentEvent{Type: "session", Data: sessionID, SessionID: sessionID})
|
||||||
@@ -204,12 +217,34 @@ func handleChat(w http.ResponseWriter, r *http.Request, a *agent, st *store) {
|
|||||||
toolCalls := []map[string]any{}
|
toolCalls := []map[string]any{}
|
||||||
var finalText string
|
var finalText string
|
||||||
|
|
||||||
|
// Incremental persistence, mirroring resumeSession's existing
|
||||||
|
// placeholder+update pattern (continue.go): insert a placeholder now,
|
||||||
|
// update the SAME row after every tool call, so whatever happened before
|
||||||
|
// an abort is never lost — only what hadn't happened yet is.
|
||||||
|
placeholder, _ := json.Marshal(map[string]any{"role": "assistant", "text": ""})
|
||||||
|
msgID, err := st.insertMessageReturningID(pctx, sessionID, "assistant", placeholder)
|
||||||
|
if err != nil {
|
||||||
|
slog.Error("nomos: chat placeholder insert failed", "session", sessionID, "error", err)
|
||||||
|
}
|
||||||
|
persist := func() {
|
||||||
|
if msgID == uuid.Nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
body, _ := json.Marshal(map[string]any{
|
||||||
|
"role": "assistant",
|
||||||
|
"text": finalText,
|
||||||
|
"tool_calls": toolCalls,
|
||||||
|
})
|
||||||
|
st.updateMessage(pctx, msgID, body)
|
||||||
|
}
|
||||||
|
|
||||||
a.chat(ctx, sessionID, req.Message, func(ev agentEvent) {
|
a.chat(ctx, sessionID, req.Message, func(ev agentEvent) {
|
||||||
if ev.Type == "tool_use" || ev.Type == "tool_result" {
|
if ev.Type == "tool_use" || ev.Type == "tool_result" {
|
||||||
if m, ok := ev.Data.(map[string]any); ok {
|
if m, ok := ev.Data.(map[string]any); ok {
|
||||||
m["type"] = ev.Type
|
m["type"] = ev.Type
|
||||||
toolCalls = append(toolCalls, m)
|
toolCalls = append(toolCalls, m)
|
||||||
}
|
}
|
||||||
|
persist() // live: survives even if the client disconnects right after
|
||||||
}
|
}
|
||||||
if ev.Type == "text" {
|
if ev.Type == "text" {
|
||||||
finalText, _ = ev.Data.(string)
|
finalText, _ = ev.Data.(string)
|
||||||
@@ -217,19 +252,14 @@ func handleChat(w http.ResponseWriter, r *http.Request, a *agent, st *store) {
|
|||||||
sseEvent(w, flusher, ev)
|
sseEvent(w, flusher, ev)
|
||||||
})
|
})
|
||||||
|
|
||||||
assistantMsg, _ := json.Marshal(map[string]any{
|
persist() // final state — same row, updated one last time with the concluding text
|
||||||
"role": "assistant",
|
|
||||||
"text": finalText,
|
|
||||||
"tool_calls": toolCalls,
|
|
||||||
})
|
|
||||||
st.saveMessage(ctx, sessionID, "assistant", assistantMsg)
|
|
||||||
|
|
||||||
// Generate a meaningful title from the assistant's first answer
|
// Generate a meaningful title from the assistant's first answer
|
||||||
// instead of reusing the raw user message for every session.
|
// instead of reusing the raw user message for every session.
|
||||||
if finalText != "" && sessionID != "ephemeral" {
|
if finalText != "" && sessionID != "ephemeral" {
|
||||||
title := truncate(finalText, 80)
|
title := truncate(finalText, 80)
|
||||||
if title != "" {
|
if title != "" {
|
||||||
st.updateSessionTitle(ctx, sessionID, title)
|
st.updateSessionTitle(pctx, sessionID, title)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user