From 34571402dbea84f8093a65964a9ad2ecd29f6a2c Mon Sep 17 00:00:00 2001 From: Shenghao Hu Date: Mon, 17 Aug 2026 14:04:09 +0900 Subject: [PATCH 1/4] feat: add conversation context actions Add complete-turn session forks, including direct cross-agent destinations, and ephemeral tool-free side chats behind a capability gate. Keep reply annotations model-visible while persisting separate display metadata so sent queries retain their annotation attachment after reload and retry. Verification: go test ./... --- docs/cloud-contract-surface.md | 20 ++ docs/desktop-wire-fixtures/README.md | 6 + .../http_get.status.response.json | 3 +- internal/daemon/client.go | 5 + internal/daemon/conversation_context.go | 193 ++++++++++++++++++ internal/daemon/conversation_context_test.go | 180 ++++++++++++++++ internal/daemon/conversation_reply.go | 127 ++++++++++++ internal/daemon/conversation_reply_test.go | 112 ++++++++++ internal/daemon/runner.go | 82 +++++--- internal/daemon/server.go | 2 + internal/session/fork_test.go | 151 ++++++++++++++ internal/session/manager.go | 125 ++++++++++++ internal/session/store.go | 17 +- 13 files changed, 988 insertions(+), 35 deletions(-) create mode 100644 internal/daemon/conversation_context.go create mode 100644 internal/daemon/conversation_context_test.go create mode 100644 internal/daemon/conversation_reply.go create mode 100644 internal/daemon/conversation_reply_test.go create mode 100644 internal/session/fork_test.go diff --git a/docs/cloud-contract-surface.md b/docs/cloud-contract-surface.md index 4a31b8bd..5f504d48 100644 --- a/docs/cloud-contract-surface.md +++ b/docs/cloud-contract-surface.md @@ -162,6 +162,26 @@ Routes the kocoro agent itself calls additionally need a matching reference unde `internal/skills/bundled/skills/kocoro/references/`; see the Doc Co-Maintenance section of `CLAUDE.md`. +Conversation context actions are Desktop-only and gated by +`conversation_context_actions_v1`: `POST /sessions/{id}/fork` copies model +history through a complete assistant turn into a normal persisted session. +Its optional `agent` identifies the source session directory; optional +`target_agent` selects the destination agent directory (use `"default"` for +Default, and omit it to branch within the source agent). +`POST /sessions/{id}/side-chat` runs against that same bounded history plus the +panel's temporary user/assistant history. Side-chat runs are ephemeral, expose +no tools, and publish no global bus events. Their implementation is +`internal/daemon/conversation_context.go`; the Desktop consumer is +`DaemonClient+ConversationContext.swift`. Neither route belongs in the bundled +Kocoro skill references because the model never calls them. + +Desktop text replies use a transient head-only `` prompt +envelope. `RunAgent` removes that envelope from the archived user message but +persists its decoded quotes and comments in the parallel +`message_meta[].conversation_annotations` display metadata. The metadata never +enters `HistoryForLoop`; it lets Desktop reconstruct the compact annotation +attachment immediately and after reload without exposing model-only markup. + ### Event payload shapes `tool_status`, `approval_request`, `approval_request.flags`, and diff --git a/docs/desktop-wire-fixtures/README.md b/docs/desktop-wire-fixtures/README.md index 0a90c9ed..b04c40a9 100644 --- a/docs/desktop-wire-fixtures/README.md +++ b/docs/desktop-wire-fixtures/README.md @@ -150,6 +150,12 @@ codecs. | `http_get.session.response.json` | `server.go handleGetSession` (`GET /sessions/{id}`) | lossless default session detail. Top-level `messages` remain the archive; `compaction_checkpoint` additively exposes the durable model-live state and its exclusive raw archive index. No new capability token is required: clients that ignore the additive field retain the existing archive semantics; consumers that opt into live-context metrics read the checkpoint explicitly. | | `http_get.session.remote_timeline.response.json` | `server.go handleGetSession` (`GET /sessions/{id}?view=remote_timeline`) | capability-gated mobile projection of the archive. Returns a byte-bounded newest page with aligned `messages` / `message_meta`, absolute `start_index`, opaque `next_cursor`, `has_more`, and explicit `omitted_content_count`; it projects from top-level archived `messages`, not the model-live checkpoint. | +`conversation_context_actions_v1` covers two Desktop-only POST routes without +golden response files: `/sessions/{id}/fork` and `/sessions/{id}/side-chat`. +Their request/response, complete-turn boundary, tool-free side-chat, and +non-persistence contracts are exercised directly in +`internal/daemon/conversation_context_test.go`. + ### Quick-panel surfaces (POST request bodies + error responses) `POST /message` optionally accepts `idempotency_key` together with a diff --git a/docs/desktop-wire-fixtures/http_get.status.response.json b/docs/desktop-wire-fixtures/http_get.status.response.json index bb18f9f8..06b3a17b 100644 --- a/docs/desktop-wire-fixtures/http_get.status.response.json +++ b/docs/desktop-wire-fixtures/http_get.status.response.json @@ -50,7 +50,8 @@ "agent_service_tier_v1", "web_search_usage_v1", "skill_install_recommendation_v1", - "work_plan_v1" + "work_plan_v1", + "conversation_context_actions_v1" ], "memory": { "provider": "disabled", diff --git a/internal/daemon/client.go b/internal/daemon/client.go index f546b0c7..cf25021a 100644 --- a/internal/daemon/client.go +++ b/internal/daemon/client.go @@ -321,6 +321,10 @@ const ( // tool, Session.work_plan on GET /sessions/{id}, and the work_plan.updated // bus event (emitted only after the snapshot's durable save). CapWorkPlanV1 = "work_plan_v1" + // CapConversationContextActionsV1 advertises the Desktop-only local + // conversation context surface: complete-turn session forks and ephemeral, + // tool-free side chats seeded from a source transcript. + CapConversationContextActionsV1 = "conversation_context_actions_v1" ) var Capabilities = []string{ @@ -371,6 +375,7 @@ var Capabilities = []string{ CapWebSearchUsageV1, CapSkillInstallRecommendationV1, CapWorkPlanV1, + CapConversationContextActionsV1, } // envelopeSenderFn lets tests substitute sendEnvelope without standing up a diff --git a/internal/daemon/conversation_context.go b/internal/daemon/conversation_context.go new file mode 100644 index 00000000..6e625efc --- /dev/null +++ b/internal/daemon/conversation_context.go @@ -0,0 +1,193 @@ +package daemon + +import ( + "errors" + "fmt" + "net/http" + "os" + "strings" + + "github.com/Kocoro-lab/ShanClaw/internal/agents" + "github.com/Kocoro-lab/ShanClaw/internal/client" + "github.com/Kocoro-lab/ShanClaw/internal/session" +) + +const maxSideChatHistoryMessages = 100 + +type forkSessionRequest struct { + Agent string `json:"agent,omitempty"` + TargetAgent *string `json:"target_agent,omitempty"` + MessageIndex int `json:"message_index"` +} + +type forkSessionResponse struct { + SessionID string `json:"session_id"` + Title string `json:"title"` +} + +type sideChatHistoryMessage struct { + Role string `json:"role"` + Content string `json:"content"` +} + +type sideChatRequest struct { + Agent string `json:"agent,omitempty"` + MessageIndex int `json:"message_index"` + Text string `json:"text"` + History []sideChatHistoryMessage `json:"history,omitempty"` +} + +func (s *Server) handleForkSession(w http.ResponseWriter, r *http.Request) { + if s.deps == nil || s.deps.SessionCache == nil { + writeError(w, http.StatusInternalServerError, "daemon deps not configured") + return + } + id, body, ok := s.decodeConversationContextRequest(w, r) + if !ok { + return + } + sourceManager := s.deps.SessionCache.GetOrCreateManager(s.deps.SessionCache.SessionsDir(body.Agent)) + targetAgent := body.Agent + if body.TargetAgent != nil { + targetAgent = *body.TargetAgent + } + targetManager := s.deps.SessionCache.GetOrCreateManager(s.deps.SessionCache.SessionsDir(targetAgent)) + fork, err := sourceManager.ForkSessionInto(id, body.MessageIndex, targetManager) + if err != nil { + writeConversationContextError(w, id, err) + return + } + writeJSON(w, http.StatusCreated, forkSessionResponse{SessionID: fork.ID, Title: fork.Title}) +} + +func (s *Server) handleSideChat(w http.ResponseWriter, r *http.Request) { + if s.deps == nil || s.deps.SessionCache == nil { + writeError(w, http.StatusInternalServerError, "daemon deps not configured") + return + } + id := r.PathValue("id") + if err := ValidateSessionID(id); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + var body sideChatRequest + if !decodeBody(w, r, &body) { + return + } + if body.Agent == "default" { + body.Agent = "" + } + if body.Agent != "" { + if err := agents.ValidateAgentName(body.Agent); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + } + if strings.TrimSpace(body.Text) == "" { + writeError(w, http.StatusBadRequest, "text is required") + return + } + if len(body.History) > maxSideChatHistoryMessages { + writeError(w, http.StatusBadRequest, "side chat history is too long") + return + } + mgr := s.deps.SessionCache.GetOrCreateManager(s.deps.SessionCache.SessionsDir(body.Agent)) + source, err := mgr.Load(id) + if err != nil { + writeConversationContextError(w, id, err) + return + } + index := body.MessageIndex + if index == 0 { + index = len(source.Messages) + } + history, err := mgr.CopyHistoryThrough(id, index) + if err != nil { + writeConversationContextError(w, id, err) + return + } + for _, item := range body.History { + role := strings.TrimSpace(item.Role) + content := strings.TrimSpace(item.Content) + if (role != "user" && role != "assistant") || content == "" { + writeError(w, http.StatusBadRequest, "side chat history must contain non-empty user/assistant messages") + return + } + history = append(history, client.Message{Role: role, Content: client.NewTextContent(content)}) + } + req := RunAgentRequest{ + Text: body.Text, + Agent: body.Agent, + Source: "desktop", + CWD: source.CWD, + ProjectID: source.ProjectID, + NewSession: true, + Ephemeral: true, + BypassRouting: true, + SessionHistory: history, + DisableTools: true, + SuppressBusEvents: true, + StickyContext: "This is a temporary side conversation. Answer from the provided conversation context. Do not modify or claim to modify the primary conversation.", + } + if err := req.Validate(); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + if strings.Contains(r.Header.Get("Accept"), "text/event-stream") { + s.handleMessageSSE(w, r, req) + return + } + handler := &httpEventHandler{} + result, err := RunAgent(r.Context(), s.deps, req, handler) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + writeJSON(w, http.StatusOK, result) +} + +func (s *Server) decodeConversationContextRequest(w http.ResponseWriter, r *http.Request) (string, forkSessionRequest, bool) { + id := r.PathValue("id") + if err := ValidateSessionID(id); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return "", forkSessionRequest{}, false + } + var body forkSessionRequest + if !decodeBody(w, r, &body) { + return "", forkSessionRequest{}, false + } + if body.Agent == "default" { + body.Agent = "" + } + if body.Agent != "" { + if err := agents.ValidateAgentName(body.Agent); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return "", forkSessionRequest{}, false + } + } + if body.TargetAgent != nil { + target := *body.TargetAgent + if target == "default" { + target = "" + } + if target != "" { + if err := agents.ValidateAgentName(target); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return "", forkSessionRequest{}, false + } + } + body.TargetAgent = &target + } + return id, body, true +} + +func writeConversationContextError(w http.ResponseWriter, id string, err error) { + switch { + case errors.Is(err, os.ErrNotExist): + writeError(w, http.StatusNotFound, fmt.Sprintf("session %q not found", id)) + case errors.Is(err, session.ErrMessageIndexOutOfRange), errors.Is(err, session.ErrIncompleteTurnBoundary): + writeError(w, http.StatusBadRequest, err.Error()) + default: + writeError(w, http.StatusInternalServerError, err.Error()) + } +} diff --git a/internal/daemon/conversation_context_test.go b/internal/daemon/conversation_context_test.go new file mode 100644 index 00000000..b8566fcd --- /dev/null +++ b/internal/daemon/conversation_context_test.go @@ -0,0 +1,180 @@ +package daemon + +import ( + "bytes" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "os" + "strings" + "testing" + + "github.com/Kocoro-lab/ShanClaw/internal/client" +) + +func TestForkSessionEndpointCreatesOrdinarySession(t *testing.T) { + shannonDir := t.TempDir() + deps := &ServerDeps{ShannonDir: shannonDir, SessionCache: NewSessionCache(shannonDir)} + server := NewServer(0, nil, deps, "test") + mgr := deps.SessionCache.GetOrCreateManager(deps.SessionCache.SessionsDir("")) + source := mgr.NewSessionWithID("source-session-http") + source.Title = "Source title" + source.CWD = "/tmp/project" + source.Messages = []client.Message{ + {Role: "user", Content: client.NewTextContent("question")}, + {Role: "assistant", Content: client.NewTextContent("answer")}, + } + if err := mgr.Save(); err != nil { + t.Fatal(err) + } + + body := bytes.NewBufferString(`{"message_index":2}`) + req := httptest.NewRequest(http.MethodPost, "/sessions/source-session-http/fork", body) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + server.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusCreated { + t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String()) + } + var response forkSessionResponse + if err := json.NewDecoder(rec.Body).Decode(&response); err != nil { + t.Fatal(err) + } + if response.SessionID == "" || response.Title != source.Title { + t.Fatalf("response = %#v", response) + } + if _, err := mgr.Load(response.SessionID); err != nil { + t.Fatalf("fork not loadable: %v", err) + } +} + +func TestForkSessionEndpointCanTargetAnotherAgent(t *testing.T) { + shannonDir := t.TempDir() + deps := &ServerDeps{ShannonDir: shannonDir, SessionCache: NewSessionCache(shannonDir)} + server := NewServer(0, nil, deps, "test") + sourceManager := deps.SessionCache.GetOrCreateManager(deps.SessionCache.SessionsDir("research")) + source := sourceManager.NewSessionWithID("source-session-cross-agent") + source.Title = "Cross-agent branch" + source.Messages = []client.Message{ + {Role: "user", Content: client.NewTextContent("question")}, + {Role: "assistant", Content: client.NewTextContent("answer")}, + } + if err := sourceManager.Save(); err != nil { + t.Fatal(err) + } + + body := bytes.NewBufferString(`{"message_index":2,"agent":"research","target_agent":"investment"}`) + req := httptest.NewRequest(http.MethodPost, "/sessions/source-session-cross-agent/fork", body) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + server.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusCreated { + t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String()) + } + var response forkSessionResponse + if err := json.NewDecoder(rec.Body).Decode(&response); err != nil { + t.Fatal(err) + } + targetManager := deps.SessionCache.GetOrCreateManager(deps.SessionCache.SessionsDir("investment")) + if _, err := targetManager.Load(response.SessionID); err != nil { + t.Fatalf("target agent cannot load fork: %v", err) + } + if _, err := sourceManager.Load(response.SessionID); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("fork persisted under source agent: %v", err) + } +} + +func TestForkSessionEndpointRejectsIncompleteTurn(t *testing.T) { + shannonDir := t.TempDir() + deps := &ServerDeps{ShannonDir: shannonDir, SessionCache: NewSessionCache(shannonDir)} + server := NewServer(0, nil, deps, "test") + mgr := deps.SessionCache.GetOrCreateManager(deps.SessionCache.SessionsDir("")) + source := mgr.NewSessionWithID("source-session-tool") + source.Messages = []client.Message{ + {Role: "user", Content: client.NewTextContent("question")}, + {Role: "assistant", Content: client.NewBlockContent([]client.ContentBlock{{Type: "tool_use", ID: "call-1", Name: "read_file"}})}, + } + if err := mgr.Save(); err != nil { + t.Fatal(err) + } + + req := httptest.NewRequest(http.MethodPost, "/sessions/source-session-tool/fork", bytes.NewBufferString(`{"message_index":2}`)) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + server.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String()) + } +} + +func TestConversationContextCapabilityAdvertised(t *testing.T) { + for _, capability := range Capabilities { + if capability == CapConversationContextActionsV1 { + return + } + } + t.Fatalf("Capabilities missing %q", CapConversationContextActionsV1) +} + +func TestSideChatEndpointUsesReadOnlyEphemeralContext(t *testing.T) { + gateway := &fakeGatewayBackend{reply: "focused answer"} + gatewayServer := httptest.NewServer(gateway.handler()) + defer gatewayServer.Close() + + deps := runAgentContractTestDeps(t, gatewayServer.URL) + defer deps.SessionCache.CloseAll() + mgr := deps.SessionCache.GetOrCreateManager(deps.SessionCache.SessionsDir("")) + source := mgr.NewSessionWithID("side-chat-source") + source.CWD = t.TempDir() + source.Messages = []client.Message{ + {Role: "user", Content: client.NewTextContent("original question")}, + {Role: "assistant", Content: client.NewTextContent("original answer")}, + } + if err := mgr.Save(); err != nil { + t.Fatal(err) + } + + server := NewServer(0, nil, deps, "test") + events := server.EventBus().Subscribe() + body := `{"message_index":2,"text":"explain this","history":[{"role":"user","content":"side follow-up"},{"role":"assistant","content":"side response"}]}` + req := httptest.NewRequest(http.MethodPost, "/sessions/side-chat-source/side-chat", bytes.NewBufferString(body)) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + server.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String()) + } + select { + case event := <-events: + t.Fatalf("side chat published global event %q", event.Type) + default: + } + + requests := gateway.requests() + if len(requests) == 0 { + t.Fatal("gateway captured no side-chat request") + } + var transcript strings.Builder + for _, request := range requests { + if len(request.Tools) != 0 { + t.Fatalf("side chat exposed %d tools", len(request.Tools)) + } + for _, message := range request.Messages { + transcript.WriteString(message.Content.Text()) + transcript.WriteByte('\n') + } + } + for _, want := range []string{"original question", "original answer", "side follow-up", "side response", "explain this"} { + if !strings.Contains(transcript.String(), want) { + t.Fatalf("gateway context omitted %q: %s", want, transcript.String()) + } + } + summaries, err := mgr.List() + if err != nil { + t.Fatal(err) + } + if len(summaries) != 1 || summaries[0].ID != source.ID { + t.Fatalf("side chat persisted an extra session: %#v", summaries) + } +} diff --git a/internal/daemon/conversation_reply.go b/internal/daemon/conversation_reply.go new file mode 100644 index 00000000..414e331b --- /dev/null +++ b/internal/daemon/conversation_reply.go @@ -0,0 +1,127 @@ +package daemon + +import ( + "encoding/xml" + "strings" + + "github.com/Kocoro-lab/ShanClaw/internal/client" + "github.com/Kocoro-lab/ShanClaw/internal/session" +) + +const ( + conversationRepliesOpen = "" + conversationRepliesClose = "" + maxConversationAnnotations = 100 + maxConversationAnnotationQuoteRunes = 8_000 + maxConversationAnnotationCommentRunes = 2_000 +) + +type conversationReplyEnvelope struct { + Replies []struct { + Quote string `xml:"quote"` + Comment string `xml:"comment"` + } `xml:"reply"` +} + +// splitConversationReplyPrompt separates Desktop's model-only reply envelope +// from the user-visible prompt. The envelope is accepted only at the head of a +// message; ordinary user text containing the same tags remains ordinary text. +func splitConversationReplyPrompt(text string) (visible string, envelope string) { + if !strings.HasPrefix(text, conversationRepliesOpen) { + return text, "" + } + end := strings.Index(text, conversationRepliesClose) + if end < 0 { + return text, "" + } + end += len(conversationRepliesClose) + return strings.TrimSpace(text[end:]), text[:end] +} + +func restoreConversationReplyPrompt(visible, envelope string) string { + if envelope == "" { + return visible + } + if visible == "" { + return envelope + } + return envelope + "\n\n" + visible +} + +func conversationReplyAnnotations(envelope string) []session.ConversationAnnotation { + if envelope == "" { + return nil + } + var parsed conversationReplyEnvelope + if err := xml.Unmarshal([]byte(envelope), &parsed); err != nil { + return nil + } + limit := len(parsed.Replies) + if limit > maxConversationAnnotations { + limit = maxConversationAnnotations + } + annotations := make([]session.ConversationAnnotation, 0, limit) + for _, reply := range parsed.Replies[:limit] { + quote := boundedConversationAnnotationText(reply.Quote, maxConversationAnnotationQuoteRunes) + comment := boundedConversationAnnotationText(reply.Comment, maxConversationAnnotationCommentRunes) + if quote == "" && comment == "" { + continue + } + annotations = append(annotations, session.ConversationAnnotation{ + SelectedText: quote, + Comment: comment, + }) + } + return annotations +} + +func boundedConversationAnnotationText(text string, maxRunes int) string { + runes := []rune(text) + if len(runes) <= maxRunes { + return text + } + return string(runes[:maxRunes]) +} + +// visibleConversationReplyText removes every well-formed model-only envelope +// from a combined prompt. Multiple envelopes occur when injected follow-ups +// are drained into one run. Persisted transcript, titles, previews, and global +// events must contain only the user's visible text. +func visibleConversationReplyText(text string) string { + removed := false + for { + start := strings.Index(text, conversationRepliesOpen) + if start < 0 { + if !removed { + return text + } + return strings.TrimSpace(text) + } + relativeEnd := strings.Index(text[start:], conversationRepliesClose) + if relativeEnd < 0 { + return strings.TrimSpace(text) + } + end := start + relativeEnd + len(conversationRepliesClose) + text = text[:start] + text[end:] + removed = true + } +} + +func visibleConversationReplyMessage(message client.Message) client.Message { + if message.Role != "user" { + return message + } + if !message.Content.HasBlocks() { + message.Content = client.NewTextContent(visibleConversationReplyText(message.Content.Text())) + return message + } + blocks := append([]client.ContentBlock(nil), message.Content.Blocks()...) + for index := range blocks { + if blocks[index].Type == "text" { + blocks[index].Text = visibleConversationReplyText(blocks[index].Text) + break + } + } + message.Content = client.NewBlockContent(blocks) + return message +} diff --git a/internal/daemon/conversation_reply_test.go b/internal/daemon/conversation_reply_test.go new file mode 100644 index 00000000..b5dbbf3c --- /dev/null +++ b/internal/daemon/conversation_reply_test.go @@ -0,0 +1,112 @@ +package daemon + +import ( + "context" + "net/http/httptest" + "strings" + "testing" + + "github.com/Kocoro-lab/ShanClaw/internal/client" + "github.com/Kocoro-lab/ShanClaw/internal/session" +) + +func TestConversationReplyPromptSeparatesModelContextFromVisibleText(t *testing.T) { + raw := "\nanswerclarify\n\n\n/inspect target" + visible, envelope := splitConversationReplyPrompt(raw) + if visible != "/inspect target" { + t.Fatalf("visible = %q", visible) + } + if got := restoreConversationReplyPrompt("expanded command", envelope); got != envelope+"\n\nexpanded command" { + t.Fatalf("restored = %q", got) + } +} + +func TestConversationReplyAnnotationsDecodeEscapingAndBounds(t *testing.T) { + envelope := `price < budgetcheck & compare` + got := conversationReplyAnnotations(envelope) + want := []session.ConversationAnnotation{{SelectedText: "price < budget", Comment: "check & compare"}} + if len(got) != len(want) || got[0] != want[0] { + t.Fatalf("annotations = %#v, want %#v", got, want) + } + if malformed := conversationReplyAnnotations(""); malformed != nil { + t.Fatalf("malformed annotations = %#v", malformed) + } +} + +func TestVisibleConversationReplyTextCleansInjectedMessages(t *testing.T) { + raw := "queued\nsecret context\nnext" + if got := visibleConversationReplyText(raw); got != "queued\n\nnext" { + t.Fatalf("visible = %q", got) + } + message := client.Message{Role: "user", Content: client.NewBlockContent([]client.ContentBlock{ + {Type: "text", Text: raw}, + {Type: "image", Source: &client.ImageSource{Type: "base64", MediaType: "image/png", Data: "abc"}}, + })} + clean := visibleConversationReplyMessage(message) + if clean.Content.Blocks()[0].Text != "queued\n\nnext" || clean.Content.Blocks()[1].Type != "image" { + t.Fatalf("cleaned blocks = %#v", clean.Content.Blocks()) + } +} + +func TestConversationReplyTagsInsideOrdinaryTextAreNotSplit(t *testing.T) { + raw := "Explain literally" + visible, envelope := splitConversationReplyPrompt(raw) + if visible != raw || envelope != "" { + t.Fatalf("ordinary text changed: visible=%q envelope=%q", visible, envelope) + } + if got := visibleConversationReplyText(" ordinary text "); got != " ordinary text " { + t.Fatalf("ordinary visible text changed: %q", got) + } +} + +func TestConversationReplyContextReachesModelButNotPersistedSession(t *testing.T) { + gateway := &fakeGatewayBackend{reply: "done"} + server := httptest.NewServer(gateway.handler()) + defer server.Close() + deps := runAgentContractTestDeps(t, server.URL) + defer deps.SessionCache.CloseAll() + + raw := "answerclarify\n\nvisible request" + result, err := RunAgent(context.Background(), deps, RunAgentRequest{ + Text: raw, + Source: "desktop", + Channel: "kocoro", + }, nullEventHandler{}) + if err != nil { + t.Fatal(err) + } + requests := gateway.requests() + var modelInput strings.Builder + for _, request := range requests { + for _, message := range request.Messages { + modelInput.WriteString(message.Content.Text()) + } + } + if !strings.Contains(modelInput.String(), "clarify") { + t.Fatalf("model input omitted reply context: %s", modelInput.String()) + } + + mgr := deps.SessionCache.GetOrCreateManager(deps.SessionCache.SessionsDir("")) + persisted, err := mgr.Load(result.SessionID) + if err != nil { + t.Fatal(err) + } + if strings.Contains(persisted.Title, conversationRepliesOpen) { + t.Fatalf("persisted title = %q", persisted.Title) + } + if len(persisted.Messages) == 0 || persisted.Messages[0].Content.Text() != "visible request" { + t.Fatalf("persisted visible request = %#v", persisted.Messages) + } + if len(persisted.MessageMeta) == 0 { + t.Fatal("persisted message metadata is empty") + } + wantAnnotations := []session.ConversationAnnotation{{SelectedText: "answer", Comment: "clarify"}} + if got := persisted.MessageMeta[0].ConversationAnnotations; len(got) != 1 || got[0] != wantAnnotations[0] { + t.Fatalf("persisted annotations = %#v, want %#v", got, wantAnnotations) + } + for _, message := range persisted.Messages { + if strings.Contains(message.Content.Text(), conversationRepliesOpen) { + t.Fatalf("persisted message leaked reply envelope: %q", message.Content.Text()) + } + } +} diff --git a/internal/daemon/runner.go b/internal/daemon/runner.go index 75c8b20e..f96fba30 100644 --- a/internal/daemon/runner.go +++ b/internal/daemon/runner.go @@ -98,6 +98,8 @@ type RunAgentRequest struct { ModelOverride string `json:"-"` // internal tier override; an exact model pin remains authoritative BypassRouting bool `json:"-"` // skip route lock (heartbeat runs) SessionHistory []client.Message `json:"-"` // pre-loaded history for LLM context (BypassRouting runs) + DisableTools bool `json:"-"` // internal read-only run: expose no model-callable tools + SuppressBusEvents bool `json:"-"` // internal request-scoped run: do not publish progress to global subscribers OmitHistory bool `json:"-"` // skip sess.HistoryForLoop() snapshot; LLM sees empty history. Set by scheduler for stateless schedules. StickyContext string `json:"-"` // 额外的 sticky context，注入系统提示（对用户不可见） ForegroundHint *ForegroundHint `json:"foreground_hint,omitempty"` // app the user was looking at when they summoned the quick panel; folded into StickyContext so screen-reading tools default to it @@ -1697,15 +1699,15 @@ type ServerDeps struct { RecordConfigMutation func(config.MutationRevisions) GW *client.GatewayClient Registry *agent.ToolRegistry - MCPManager *mcp.ClientManager // live MCP connections; swapped on reload - Supervisor *mcp.Supervisor // MCP health supervisor; swapped on reload + MCPManager *mcp.ClientManager // live MCP connections; swapped on reload + Supervisor *mcp.Supervisor // MCP health supervisor; swapped on reload // MCPReconnect schedules bounded retries for servers whose async connect // failed. Owned by the MCPManager generation that spawned it, so it is // swapped (and the old one stopped) alongside MCPManager on reload — // letting a superseded ladder keep firing would resurrect connections on // a manager nothing can reach. - MCPReconnect *mcp.ReconnectScheduler - Cleanup func() // closes MCP connections; swapped on reload + MCPReconnect *mcp.ReconnectScheduler + Cleanup func() // closes MCP connections; swapped on reload BaselineReg *agent.ToolRegistry // local-only tools; refreshed on reload GatewayOverlay []agent.Tool // cached gateway tools; refreshed on reload PostOverlays []agent.Tool // cloud_delegate etc.; refreshed on reload @@ -2581,7 +2583,9 @@ func RunAgent(ctx context.Context, deps *ServerDeps, req RunAgentRequest, handle return nil, fmt.Errorf("daemon not fully configured") } agentName := req.Agent - prompt := req.Text + visibleInput, replyEnvelope := splitConversationReplyPrompt(req.Text) + replyAnnotations := conversationReplyAnnotations(replyEnvelope) + prompt := visibleInput // Download remote file attachments and convert to file_ref blocks. // Attachment files must survive across turns (non-image files become @@ -2636,10 +2640,10 @@ func RunAgent(ctx context.Context, deps *ServerDeps, req RunAgentRequest, handle // re-routed here to that other agent, so two apps collapse onto one agent and // answer identically. Mirrors the same guard in cmd/daemon.go onMsg. if agentName == "" && !IsMessagingPlatform(req.Source) { - agentName, prompt = agents.ParseAgentMention(req.Text) + agentName, prompt = agents.ParseAgentMention(visibleInput) } if prompt == "" { - prompt = req.Text + prompt = visibleInput } var agentOverride *agents.Agent @@ -2652,7 +2656,7 @@ func RunAgent(ctx context.Context, deps *ServerDeps, req RunAgentRequest, handle // @mention fallback: use default agent log.Printf("daemon: agent %q not found: %v, using default", agentName, loadErr) agentName = "" - prompt = req.Text + prompt = visibleInput } else { agentOverride = a } @@ -2669,7 +2673,9 @@ func RunAgent(ctx context.Context, deps *ServerDeps, req RunAgentRequest, handle prompt = strings.ReplaceAll(content, "$ARGUMENTS", args) } } - req.Text = prompt + visiblePrompt := prompt + prompt = restoreConversationReplyPrompt(visiblePrompt, replyEnvelope) + req.Text = visiblePrompt // Recompute route key after final agent resolution. // Callers may precompute a default/source-channel key before @mention parsing. // Recomputing here avoids cross-route contamination. @@ -2853,7 +2859,8 @@ func RunAgent(ctx context.Context, deps *ServerDeps, req RunAgentRequest, handle } else { prompt = b.String() + "\n" + prompt } - req.Text = prompt + visiblePrompt = visibleConversationReplyText(prompt) + req.Text = visiblePrompt } log.Printf("daemon: drained %d mailbox msg(s) into prompt for route %q", len(pendingBatch), req.RouteKey) // Defer the durable consumed_at flag + EventQueueFlushed to @@ -3214,8 +3221,12 @@ func RunAgent(ctx context.Context, deps *ServerDeps, req RunAgentRequest, handle // Wrap the transport handler with a bus-emitting handler so every run // publishes progress events regardless of transport. See // docs/superpowers/specs/2026-04-23-event-bus-progress-coverage-design.md. - bus := &busEventHandler{deps: deps, agent: agentName} - combinedHandler := &multiHandler{handlers: []agent.EventHandler{handler, bus}} + combinedHandler := &multiHandler{handlers: []agent.EventHandler{handler}} + if !req.SuppressBusEvents { + bus := &busEventHandler{deps: deps, agent: agentName} + combinedHandler.handlers = append(combinedHandler.handlers, bus) + } + handler = combinedHandler var runEvents *runEventCollector if !req.Ephemeral { eventLog, err := sessMgr.OpenRunEventLog(sess.ID, req.RunID, req.AttemptID, cfg.Agent.RunEventRetention) @@ -3224,8 +3235,6 @@ func RunAgent(ctx context.Context, deps *ServerDeps, req RunAgentRequest, handle } runEvents = newRunEventCollector(combinedHandler, eventLog, sess.ID, req.RunID, req.AttemptID) handler = runEvents - } else { - handler = combinedHandler } // Notify handler of resolved session ID so it can include it in EventBus payloads. @@ -3320,16 +3329,17 @@ func RunAgent(ctx context.Context, deps *ServerDeps, req RunAgentRequest, handle source = "unknown" } msgID := generateMessageID() - userMsgContent := buildUserMsgContent(prompt, resolvedContent) + userMsgContent := buildUserMsgContent(visiblePrompt, resolvedContent) sess.Messages = append(sess.Messages, client.Message{Role: "user", Content: userMsgContent}, ) sess.MessageMeta = append(sess.MessageMeta, session.MessageMeta{ - Source: source, - MessageID: msgID, - Timestamp: session.TimePtr(userMsgTime), - SystemInjected: req.ResumeInterrupted || req.SystemInjected, + Source: source, + MessageID: msgID, + Timestamp: session.TimePtr(userMsgTime), + SystemInjected: req.ResumeInterrupted || req.SystemInjected, + ConversationAnnotations: replyAnnotations, }, ) preLoopUserAppended = true @@ -3346,7 +3356,7 @@ func RunAgent(ctx context.Context, deps *ServerDeps, req RunAgentRequest, handle "sender": req.Sender, "session_id": sess.ID, "message_id": msgID, - "text": prompt, + "text": visiblePrompt, }) deps.EventBus.Emit(Event{Type: EventMessageReceived, Payload: payload}) } @@ -3594,6 +3604,9 @@ func RunAgent(ctx context.Context, deps *ServerDeps, req RunAgentRequest, handle }) } defer guiWorkflow.EndTurn() + if req.DisableTools { + reg = agent.NewToolRegistry() + } loop := agent.NewAgentLoop(deps.GW, reg, runCfg.ModelTier, deps.ShannonDir, runCfg.Agent.MaxIterations, runCfg.Tools.ResultTruncation, runCfg.Tools.ArgsTruncation, @@ -4088,6 +4101,7 @@ func RunAgent(ctx context.Context, deps *ServerDeps, req RunAgentRequest, handle checkpointSource = "unknown" } turnBase := captureTurnBaseline(sess, checkpointSource, preLoopUserAppended) + turnBase.conversationAnnotations = replyAnnotations // The daemon handler implements agent.UsageProvider; extract once so // callsites pass a strongly-typed provider (or nil) to applyTurnState. var turnUsage usageProvider @@ -4353,7 +4367,7 @@ func RunAgent(ctx context.Context, deps *ServerDeps, req RunAgentRequest, handle // identically to the default agent — the smart-title upgrade replaces // this placeholder asynchronously. if sess.Title == "New session" { - sess.Title = session.Title(prompt) + sess.Title = session.Title(visiblePrompt) sess.TitleAuto = true } @@ -4373,12 +4387,16 @@ func RunAgent(ctx context.Context, deps *ServerDeps, req RunAgentRequest, handle sess.MessageMeta = sess.MessageMeta[:turnBase.metaCount] } if !preLoopUserAppended { - fallbackContent := buildUserMsgContent(prompt, resolvedContent) + fallbackContent := buildUserMsgContent(visiblePrompt, resolvedContent) sess.Messages = append(sess.Messages, client.Message{Role: "user", Content: fallbackContent}, ) sess.MessageMeta = append(sess.MessageMeta, - session.MessageMeta{Source: checkpointSource, Timestamp: session.TimePtr(userMsgTime)}, + session.MessageMeta{ + Source: checkpointSource, + Timestamp: session.TimePtr(userMsgTime), + ConversationAnnotations: replyAnnotations, + }, ) } replyTime := time.Now() @@ -5182,12 +5200,13 @@ func isSoftRunError(err error) bool { // for the accumulated turn, no matter how many times the function is // called. type turnBaseline struct { - msgCount int - metaCount int - usage session.UsageSummary // pre-turn cumulative usage; zero if sess.Usage was nil - hadUsage bool // true if sess.Usage was non-nil at baseline - source string - preLoopUser bool + msgCount int + metaCount int + usage session.UsageSummary // pre-turn cumulative usage; zero if sess.Usage was nil + hadUsage bool // true if sess.Usage was non-nil at baseline + source string + preLoopUser bool + conversationAnnotations []session.ConversationAnnotation } // captureTurnBaseline snapshots sess state at turn start so subsequent @@ -5243,8 +5262,11 @@ func applyTurnMessages(sess *session.Session, loop *agent.AgentLoop, b turnBasel if i < len(runTimestamps) && !runTimestamps[i].IsZero() { ts = runTimestamps[i] } - sess.Messages = append(sess.Messages, runMsgs[i]) + sess.Messages = append(sess.Messages, visibleConversationReplyMessage(runMsgs[i])) meta := session.MessageMeta{Source: b.source, Timestamp: session.TimePtr(ts)} + if i == startIdx && runMsgs[i].Role == "user" { + meta.ConversationAnnotations = b.conversationAnnotations + } if i < len(runInjected) && runInjected[i] { meta.SystemInjected = true } diff --git a/internal/daemon/server.go b/internal/daemon/server.go index 774c4616..9ad519e5 100644 --- a/internal/daemon/server.go +++ b/internal/daemon/server.go @@ -945,6 +945,8 @@ func (s *Server) registerRoutes(mux *http.ServeMux) { mux.HandleFunc("DELETE /sessions/{id}", s.handleDeleteSession) mux.HandleFunc("PATCH /sessions/{id}", s.handlePatchSession) mux.HandleFunc("POST /sessions/{id}/edit", s.handleEditMessage) + mux.HandleFunc("POST /sessions/{id}/fork", s.handleForkSession) + mux.HandleFunc("POST /sessions/{id}/side-chat", s.handleSideChat) mux.HandleFunc("POST /sessions/{id}/reset", s.handleResetSession) mux.HandleFunc("POST /sessions/{id}/rewind", s.handleRewind) mux.HandleFunc("GET /sessions/{id}/summary", s.handleSessionSummary) diff --git a/internal/session/fork_test.go b/internal/session/fork_test.go new file mode 100644 index 00000000..47f234ee --- /dev/null +++ b/internal/session/fork_test.go @@ -0,0 +1,151 @@ +package session + +import ( + "errors" + "os" + "path/filepath" + "testing" + "time" + + "github.com/Kocoro-lab/ShanClaw/internal/client" +) + +func TestForkSessionCopiesConversationWithoutRuntimeState(t *testing.T) { + mgr := NewManager(t.TempDir()) + source := mgr.NewSessionWithID("source-session-1") + source.Title = "Travel plan" + source.TitleAuto = true + source.CWD = "/tmp/project" + source.ProjectID = "project-1" + source.Source = "desktop" + source.Messages = []client.Message{ + mkMsg("user", "first"), + mkMsg("assistant", "answer one"), + mkMsg("user", "second"), + mkMsg("assistant", "answer two"), + } + now := time.Now() + source.MessageMeta = []MessageMeta{ + {MessageID: "u1", Timestamp: &now}, + {MessageID: "a1", Timestamp: &now}, + {MessageID: "u2", Timestamp: &now}, + {MessageID: "a2", Timestamp: &now}, + } + source.Usage = &UsageSummary{LLMCalls: 2} + source.WorkPlan = &WorkPlanSnapshot{Lifecycle: WorkPlanActive} + source.InProgress = true + source.RouteKey = "session:source-session-1" + if err := mgr.Save(); err != nil { + t.Fatal(err) + } + + fork, err := mgr.ForkSession(source.ID, 2) + if err != nil { + t.Fatal(err) + } + if fork.ID == source.ID { + t.Fatal("fork reused source id") + } + if got := len(fork.Messages); got != 2 { + t.Fatalf("fork message count = %d, want 2", got) + } + if fork.Messages[1].Content.Text() != "answer one" { + t.Fatalf("fork reply = %q", fork.Messages[1].Content.Text()) + } + if fork.Title != source.Title || fork.CWD != source.CWD || fork.ProjectID != source.ProjectID { + t.Fatalf("fork lost conversation identity: %#v", fork) + } + if fork.Usage != nil || fork.WorkPlan != nil || fork.InProgress || fork.RouteKey != "" { + t.Fatalf("fork inherited runtime state: %#v", fork) + } + loaded, err := mgr.Load(fork.ID) + if err != nil { + t.Fatalf("fork was not persisted: %v", err) + } + if len(loaded.Messages) != 2 { + t.Fatalf("persisted fork message count = %d", len(loaded.Messages)) + } +} + +func TestForkSessionIntoPersistsInTargetAgentDirectory(t *testing.T) { + root := t.TempDir() + sourceManager := NewManager(filepath.Join(root, "source")) + targetManager := NewManager(filepath.Join(root, "target")) + source := sourceManager.NewSessionWithID("source-session-cross-agent") + source.Title = "Cross-agent context" + source.Messages = []client.Message{ + mkMsg("user", "question"), + mkMsg("assistant", "answer"), + } + if err := sourceManager.Save(); err != nil { + t.Fatal(err) + } + + fork, err := sourceManager.ForkSessionInto(source.ID, 2, targetManager) + if err != nil { + t.Fatal(err) + } + if _, err := targetManager.Load(fork.ID); err != nil { + t.Fatalf("target manager could not load fork: %v", err) + } + if _, err := sourceManager.Load(fork.ID); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("fork leaked into source manager: %v", err) + } +} + +func TestCompleteTurnBoundaryRejectsToolTrajectory(t *testing.T) { + toolUse := client.Message{Role: "assistant", Content: client.NewBlockContent([]client.ContentBlock{{ + Type: "tool_use", ID: "call-1", Name: "read_file", + }})} + toolResult := client.Message{Role: "user", Content: client.NewBlockContent([]client.ContentBlock{{ + Type: "tool_result", ToolUseID: "call-1", ToolContent: "ok", + }})} + messages := []client.Message{ + mkMsg("user", "inspect"), + toolUse, + toolResult, + mkMsg("assistant", "done"), + } + for _, index := range []int{1, 2, 3} { + if err := validateCompleteTurnBoundary(messages, index); !errors.Is(err, ErrIncompleteTurnBoundary) { + t.Fatalf("boundary %d error = %v, want incomplete turn", index, err) + } + } + if err := validateCompleteTurnBoundary(messages, 4); err != nil { + t.Fatalf("final boundary rejected: %v", err) + } +} + +func TestCopyHistoryThroughDropsInjectedMessagesAndDoesNotAlias(t *testing.T) { + mgr := NewManager(t.TempDir()) + source := mgr.NewSessionWithID("source-session-2") + source.Messages = []client.Message{ + mkMsg("user", "hello"), + mkMsg("assistant", "internal guard"), + mkMsg("assistant", "answer"), + } + source.MessageMeta = []MessageMeta{ + {}, + {SystemInjected: true}, + {}, + } + if err := mgr.Save(); err != nil { + t.Fatal(err) + } + + history, err := mgr.CopyHistoryThrough(source.ID, 3) + if err != nil { + t.Fatal(err) + } + if len(history) != 2 || history[1].Content.Text() != "answer" { + t.Fatalf("copied history = %#v", history) + } + history[0] = mkMsg("user", "changed") + reloaded, err := mgr.Load(source.ID) + if err != nil { + t.Fatal(err) + } + if reloaded.Messages[0].Content.Text() != "hello" { + t.Fatal("copied history aliases the persisted source") + } +} diff --git a/internal/session/manager.go b/internal/session/manager.go index 710cf193..fd889d15 100644 --- a/internal/session/manager.go +++ b/internal/session/manager.go @@ -3,6 +3,7 @@ package session import ( "crypto/rand" "encoding/hex" + "encoding/json" "errors" "fmt" "regexp" @@ -10,6 +11,7 @@ import ( "time" "github.com/Kocoro-lab/ShanClaw/internal/agent" + "github.com/Kocoro-lab/ShanClaw/internal/client" ) // ErrMessageIndexOutOfRange is returned by TruncateMessages when the requested @@ -18,6 +20,12 @@ import ( // handleEditMessage can map each to the correct status. var ErrMessageIndexOutOfRange = errors.New("message_index out of range") +// ErrIncompleteTurnBoundary is returned when a caller tries to copy history +// from the middle of an assistant/tool trajectory. A branch or side chat must +// start after a complete assistant turn so the copied context never contains a +// tool_use without its result or an unfinished model response. +var ErrIncompleteTurnBoundary = errors.New("message_index is not a complete turn boundary") + // validSessionIDPattern restricts client-supplied session IDs to a safe shape. // Allowed: 8–80 chars of [A-Za-z0-9._-]. The cap matches generateID's // "YYYY-MM-DD-" upper bound; the character class permits both daemon-minted @@ -551,6 +559,123 @@ func (m *Manager) TruncateMessages(id string, index int) error { return m.store.Save(sess) } +// CopyHistoryThrough returns an independent, model-visible copy of a session's +// history through index. index is exclusive and must end at a complete +// assistant turn. Loop-internal injected messages are omitted. +func (m *Manager) CopyHistoryThrough(id string, index int) ([]client.Message, error) { + m.mu.Lock() + defer m.mu.Unlock() + + sess, err := m.store.Load(id) + if err != nil { + return nil, err + } + if err := validateCompleteTurnBoundary(sess.Messages, index); err != nil { + return nil, err + } + meta := sess.MessageMeta + if len(meta) > index { + meta = meta[:index] + } + history := FilterInjected(sess.Messages[:index], meta) + return cloneMessages(history) +} + +// ForkSession creates an ordinary persisted session containing the source +// transcript through index. Runtime/task state is intentionally not inherited: +// the new session keeps conversation context, CWD, project, and message +// timestamps, but starts with no active work plan, usage ledger, route binding, +// share records, or in-progress execution. +func (m *Manager) ForkSession(id string, index int) (*Session, error) { + return m.ForkSessionInto(id, index, m) +} + +// ForkSessionInto persists the fork in target. Source and target managers may +// point at different agent session directories. +func (m *Manager) ForkSessionInto(id string, index int, target *Manager) (*Session, error) { + if target == nil { + target = m + } + m.mu.Lock() + source, err := m.store.Load(id) + if err != nil { + m.mu.Unlock() + return nil, err + } + if err := validateCompleteTurnBoundary(source.Messages, index); err != nil { + m.mu.Unlock() + return nil, err + } + messages, err := cloneMessages(source.Messages[:index]) + if err != nil { + m.mu.Unlock() + return nil, err + } + metaCount := min(index, len(source.MessageMeta)) + meta := append([]MessageMeta(nil), source.MessageMeta[:metaCount]...) + now := time.Now() + fork := &Session{ + SchemaVersion: 1, + ID: generateID(), + CreatedAt: now, + Title: source.Title, + TitleAuto: source.TitleAuto, + TitleTurns: source.TitleTurns, + CWD: source.CWD, + Messages: messages, + MessageMeta: meta, + Source: source.Source, + ProjectID: source.ProjectID, + } + m.mu.Unlock() + + target.mu.Lock() + defer target.mu.Unlock() + if err := target.store.Save(fork); err != nil { + return nil, err + } + return fork, nil +} + +func validateCompleteTurnBoundary(messages []client.Message, index int) error { + if index < 1 || index > len(messages) { + return fmt.Errorf("%w: %d not in [1, %d]", ErrMessageIndexOutOfRange, index, len(messages)) + } + last := messages[index-1] + if last.Role != "assistant" || containsToolBlock(last, "tool_use") || containsToolBlock(last, client.OpenAIComputerCallType) { + return ErrIncompleteTurnBoundary + } + if index == len(messages) { + return nil + } + next := messages[index] + if next.Role != "user" || containsToolBlock(next, "tool_result") { + return ErrIncompleteTurnBoundary + } + return nil +} + +func containsToolBlock(message client.Message, blockType string) bool { + for _, block := range message.Content.Blocks() { + if block.Type == blockType { + return true + } + } + return false +} + +func cloneMessages(messages []client.Message) ([]client.Message, error) { + data, err := json.Marshal(messages) + if err != nil { + return nil, fmt.Errorf("copy session messages: %w", err) + } + var cloned []client.Message + if err := json.Unmarshal(data, &cloned); err != nil { + return nil, fmt.Errorf("copy session messages: %w", err) + } + return cloned, nil +} + // ResumeLatest loads the most recently updated session from disk. // Returns (nil, nil) if no sessions exist. func (m *Manager) ResumeLatest() (*Session, error) { diff --git a/internal/session/store.go b/internal/session/store.go index 243acc31..802abd02 100644 --- a/internal/session/store.go +++ b/internal/session/store.go @@ -25,10 +25,19 @@ func TimePtr(t time.Time) *time.Time { return &t } // MessageMeta holds per-message metadata not sent to the LLM gateway. // Indexed parallel to Session.Messages. type MessageMeta struct { - Source string `json:"source,omitempty"` // "local", "slack", "line", "kocoro", "webhook", "scheduler" (legacy "shanclaw" still appears in older sessions) - MessageID string `json:"message_id,omitempty"` // stable ID for dedup (e.g. "msg-") - Timestamp *time.Time `json:"timestamp,omitempty"` // when this message was sent/received; nil = legacy (pre-timestamp) - SystemInjected bool `json:"system_injected,omitempty"` // true for guardrail/nudge messages injected by the agent loop + Source string `json:"source,omitempty"` // "local", "slack", "line", "kocoro", "webhook", "scheduler" (legacy "shanclaw" still appears in older sessions) + MessageID string `json:"message_id,omitempty"` // stable ID for dedup (e.g. "msg-") + Timestamp *time.Time `json:"timestamp,omitempty"` // when this message was sent/received; nil = legacy (pre-timestamp) + SystemInjected bool `json:"system_injected,omitempty"` // true for guardrail/nudge messages injected by the agent loop + ConversationAnnotations []ConversationAnnotation `json:"conversation_annotations,omitempty"` // user-visible reply anchors, excluded from model history +} + +// ConversationAnnotation is the durable display metadata for one text reply. +// The model receives the equivalent data through the transient reply envelope; +// this copy exists only so clients can reconstruct the sent-message attachment. +type ConversationAnnotation struct { + SelectedText string `json:"selected_text"` + Comment string `json:"comment,omitempty"` } // DeliverableReceipt is the durable, daemon-validated metadata emitted only From 3ce4d4fd9928206bad64d799622112d2132cc32f Mon Sep 17 00:00:00 2001 From: Shenghao Hu Date: Mon, 17 Aug 2026 16:08:44 +0900 Subject: [PATCH 2/4] fix: harden conversation reply persistence, validation, and fork context Reply annotations previously survived only on the message that triggered a run. A follow-up queued while the agent was busy (Desktop enqueues wire text with the full envelope whenever a run is active) reached the model, but its annotations were never decoded into MessageMeta, so the UI chips vanished on reload. Envelope stripping also matched any well-formed tag pair anywhere in a user message, eating user-authored text, and the documented 100/8000/2000 limits only bounded the persistence projection, not what the model received. Forks and side chats dropped the compaction checkpoint, re-feeding (and re-compacting) the full raw archive of compacted sessions. - Decode and persist annotations for every non-injected user message in a run: pre-run mailbox drains merge per-message envelopes (raw bytes) into one head run with merged annotations; mid-run injected turns are handled segment-aware behind the "[New message from user]" prefix (now an exported agent const). - Strip envelopes head-only and only when they parse; a delimited-but- malformed envelope stays verbatim (lossless) instead of evaporating. - Enforce the reply limits at POST /message and POST /queue on the exact envelope bytes, with stable codes conversation_replies_malformed / conversation_replies_too_many / conversation_reply_quote_too_long / conversation_reply_comment_too_long. - Carry a covering CompactionCheckpoint into ForkSessionInto (deep copy) and serve side-chat history via the new Session.HistoryThrough checkpoint+tail view; coverage past the cut is dropped and self-heals. - Validate fork target_agent / side-chat agent existence (400 agent_not_found) so a typo can no longer file sessions into an orphan directory no UI ever lists. - Contract: document the message_index raw-archive basis and the ingress limits in docs/cloud-contract-surface.md; add the http_post.session_fork.response.json wire fixture with a real-producer test and drop the fixture exemption from the README. Verification: go test ./... passes (full suite, exit 0); go test -race passes on internal/session and the new internal/daemon conversation tests. New regression coverage: drained-mailbox and mid-run-injected annotation persistence, head-only + malformed-lossless stripping, ingress limit 400s, SSE side-chat transport, compacted-source fork/side-chat, concurrent forks, unknown-agent rejection, fork response fixture. --- docs/cloud-contract-surface.md | 20 ++ docs/desktop-wire-fixtures/README.md | 7 +- .../http_post.session_fork.response.json | 4 + internal/agent/inject_types.go | 7 +- internal/daemon/conversation_context.go | 29 +++ internal/daemon/conversation_context_test.go | 167 +++++++++++- internal/daemon/conversation_reply.go | 227 +++++++++++++---- internal/daemon/conversation_reply_test.go | 237 +++++++++++++++++- internal/daemon/queue_enqueue_handler.go | 6 + internal/daemon/runner.go | 83 ++++-- internal/daemon/server.go | 6 + internal/daemon/wire_fixtures_test.go | 45 ++++ internal/session/fork_test.go | 136 ++++++++++ internal/session/manager.go | 52 ++-- internal/session/store.go | 41 ++- 15 files changed, 965 insertions(+), 102 deletions(-) create mode 100644 docs/desktop-wire-fixtures/http_post.session_fork.response.json diff --git a/docs/cloud-contract-surface.md b/docs/cloud-contract-surface.md index 5f504d48..9735be7b 100644 --- a/docs/cloud-contract-surface.md +++ b/docs/cloud-contract-surface.md @@ -175,12 +175,32 @@ no tools, and publish no global bus events. Their implementation is `DaemonClient+ConversationContext.swift`. Neither route belongs in the bundled Kocoro skill references because the model never calls them. +`message_index` on both routes is a boundary in the RAW archive index space — +the `messages` array of the session file, system-injected entries included — +equal to (index of the last included message) + 1, and it must land on a +complete assistant turn. Desktop derives it from `SessionDisplayMapper`'s +`rawIndex` (raw `enumerated()` position, injected entries skipped for display +but never renumbered), so the two sides share one basis; this was cross-checked +against the Desktop implementation on 2026-08-17. Both routes honor a +compaction checkpoint whose coverage ends at or before the boundary: the fork +carries the checkpoint (deep-copied) and side-chat feeds the model +checkpoint+tail, never the full raw archive. + Desktop text replies use a transient head-only `` prompt envelope. `RunAgent` removes that envelope from the archived user message but persists its decoded quotes and comments in the parallel `message_meta[].conversation_annotations` display metadata. The metadata never enters `HistoryForLoop`; it lets Desktop reconstruct the compact annotation attachment immediately and after reload without exposing model-only markup. +The envelope limits are enforced server-side at both Desktop run ingresses +(`POST /message`, `POST /queue`) on the exact bytes the +model would receive: ≤ 100 replies per envelope run, quotes ≤ 8,000 runes, +comments ≤ 2,000 runes. Violations are 400s with stable codes +`conversation_replies_too_many` / `conversation_reply_quote_too_long` / +`conversation_reply_comment_too_long`, and a delimited-but-unparseable +envelope is `conversation_replies_malformed`. On paths that cannot reject +(queue drain, non-Desktop sources) a malformed envelope is kept verbatim as +ordinary text — user bytes are never silently dropped. ### Event payload shapes diff --git a/docs/desktop-wire-fixtures/README.md b/docs/desktop-wire-fixtures/README.md index b04c40a9..d17746c5 100644 --- a/docs/desktop-wire-fixtures/README.md +++ b/docs/desktop-wire-fixtures/README.md @@ -149,12 +149,7 @@ codecs. | `http_get.sessions.schedule.response.json` | `server.go handleSessions` (`GET /sessions?schedule_id=`) | exact scheduled-task session filter. Each matching row carries persistent `schedule_id`; deleting the schedule configuration does not delete or rewrite the session. | | `http_get.session.response.json` | `server.go handleGetSession` (`GET /sessions/{id}`) | lossless default session detail. Top-level `messages` remain the archive; `compaction_checkpoint` additively exposes the durable model-live state and its exclusive raw archive index. No new capability token is required: clients that ignore the additive field retain the existing archive semantics; consumers that opt into live-context metrics read the checkpoint explicitly. | | `http_get.session.remote_timeline.response.json` | `server.go handleGetSession` (`GET /sessions/{id}?view=remote_timeline`) | capability-gated mobile projection of the archive. Returns a byte-bounded newest page with aligned `messages` / `message_meta`, absolute `start_index`, opaque `next_cursor`, `has_more`, and explicit `omitted_content_count`; it projects from top-level archived `messages`, not the model-live checkpoint. | - -`conversation_context_actions_v1` covers two Desktop-only POST routes without -golden response files: `/sessions/{id}/fork` and `/sessions/{id}/side-chat`. -Their request/response, complete-turn boundary, tool-free side-chat, and -non-persistence contracts are exercised directly in -`internal/daemon/conversation_context_test.go`. +| `http_post.session_fork.response.json` | `conversation_context.go handleForkSession` (`POST /sessions/{id}/fork`) | fork response for `conversation_context_actions_v1`. `session_id` is the freshly generated fork id (shape-asserted, normalized in the test); `title` snapshots the source session's title. The side-chat route has no fixture of its own: its non-SSE response is a `RunAgentResult` and its SSE stream reuses the already-pinned per-request frames; the boundary, tool-free, and non-persistence contracts are exercised in `internal/daemon/conversation_context_test.go`. | ### Quick-panel surfaces (POST request bodies + error responses) diff --git a/docs/desktop-wire-fixtures/http_post.session_fork.response.json b/docs/desktop-wire-fixtures/http_post.session_fork.response.json new file mode 100644 index 00000000..33af0b04 --- /dev/null +++ b/docs/desktop-wire-fixtures/http_post.session_fork.response.json @@ -0,0 +1,4 @@ +{ + "session_id": "2026-08-17-a1b2c3d4e5f6", + "title": "Source title" +} diff --git a/internal/agent/inject_types.go b/internal/agent/inject_types.go index dd4d4c53..e2438479 100644 --- a/internal/agent/inject_types.go +++ b/internal/agent/inject_types.go @@ -45,6 +45,11 @@ drain: return out } +// InjectedUserMessagePrefix marks a text-only injected-turn user message. +// Exported because the daemon's persistence path keys off it to locate each +// drained follow-up's head inside the joined text. +const InjectedUserMessagePrefix = "[New message from user]\n" + // buildInjectedUserMessage converts drained InjectedMessages into a single // user-role Message ready to append. Returns (zero, false) when input is // empty. When NO Files are present, emits a plain text content (preserving @@ -70,7 +75,7 @@ func buildInjectedUserMessage(drained []InjectedMessage) (client.Message, bool) // is byte-identical to the pre-refactor inject behavior. return client.Message{ Role: "user", - Content: client.NewTextContent("[New message from user]\n" + combinedText), + Content: client.NewTextContent(InjectedUserMessagePrefix + combinedText), }, true } blocks := make([]client.ContentBlock, 0, 1+len(files)) diff --git a/internal/daemon/conversation_context.go b/internal/daemon/conversation_context.go index 6e625efc..e6aba964 100644 --- a/internal/daemon/conversation_context.go +++ b/internal/daemon/conversation_context.go @@ -5,6 +5,7 @@ import ( "fmt" "net/http" "os" + "path/filepath" "strings" "github.com/Kocoro-lab/ShanClaw/internal/agents" @@ -51,6 +52,13 @@ func (s *Server) handleForkSession(w http.ResponseWriter, r *http.Request) { if body.TargetAgent != nil { targetAgent = *body.TargetAgent } + // GetOrCreateManager creates the target session directory, so a typo'd + // agent would silently file the fork where no UI ever lists it. + if !s.conversationAgentExists(targetAgent) { + writeErrorCode(w, http.StatusBadRequest, "agent_not_found", + fmt.Sprintf("agent not found: %s", targetAgent)) + return + } targetManager := s.deps.SessionCache.GetOrCreateManager(s.deps.SessionCache.SessionsDir(targetAgent)) fork, err := sourceManager.ForkSessionInto(id, body.MessageIndex, targetManager) if err != nil { @@ -83,6 +91,13 @@ func (s *Server) handleSideChat(w http.ResponseWriter, r *http.Request) { return } } + // The ephemeral run executes AS this agent; failing here beats a late + // "agent not found" 500 out of RunAgent. + if !s.conversationAgentExists(body.Agent) { + writeErrorCode(w, http.StatusBadRequest, "agent_not_found", + fmt.Sprintf("agent not found: %s", body.Agent)) + return + } if strings.TrimSpace(body.Text) == "" { writeError(w, http.StatusBadRequest, "text is required") return @@ -181,6 +196,20 @@ func (s *Server) decodeConversationContextRequest(w http.ResponseWriter, r *http return id, body, true } +// conversationAgentExists reports whether a named agent is defined (user dir +// or _builtin fallback, mirroring agents.LoadAgent's resolution). The empty +// name is the default agent, which always exists. +func (s *Server) conversationAgentExists(name string) bool { + if name == "" { + return true + } + if _, err := os.Stat(filepath.Join(s.deps.AgentsDir, name, "AGENT.md")); err == nil { + return true + } + _, err := os.Stat(filepath.Join(s.deps.AgentsDir, "_builtin", name, "AGENT.md")) + return err == nil +} + func writeConversationContextError(w http.ResponseWriter, id string, err error) { switch { case errors.Is(err, os.ErrNotExist): diff --git a/internal/daemon/conversation_context_test.go b/internal/daemon/conversation_context_test.go index b8566fcd..a6925b57 100644 --- a/internal/daemon/conversation_context_test.go +++ b/internal/daemon/conversation_context_test.go @@ -7,10 +7,13 @@ import ( "net/http" "net/http/httptest" "os" + "path/filepath" "strings" "testing" "github.com/Kocoro-lab/ShanClaw/internal/client" + ctxwin "github.com/Kocoro-lab/ShanClaw/internal/context" + "github.com/Kocoro-lab/ShanClaw/internal/session" ) func TestForkSessionEndpointCreatesOrdinarySession(t *testing.T) { @@ -49,10 +52,26 @@ func TestForkSessionEndpointCreatesOrdinarySession(t *testing.T) { } } +// writeTestAgentDefinition creates the minimal on-disk agent (AGENT.md) that +// conversationAgentExists resolves. +func writeTestAgentDefinition(t *testing.T, agentsDir, name string) { + t.Helper() + dir := filepath.Join(agentsDir, name) + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, "AGENT.md"), []byte("test agent"), 0o644); err != nil { + t.Fatal(err) + } +} + func TestForkSessionEndpointCanTargetAnotherAgent(t *testing.T) { shannonDir := t.TempDir() - deps := &ServerDeps{ShannonDir: shannonDir, SessionCache: NewSessionCache(shannonDir)} + agentsDir := filepath.Join(shannonDir, "agents") + deps := &ServerDeps{ShannonDir: shannonDir, AgentsDir: agentsDir, SessionCache: NewSessionCache(shannonDir)} server := NewServer(0, nil, deps, "test") + writeTestAgentDefinition(t, agentsDir, "research") + writeTestAgentDefinition(t, agentsDir, "investment") sourceManager := deps.SessionCache.GetOrCreateManager(deps.SessionCache.SessionsDir("research")) source := sourceManager.NewSessionWithID("source-session-cross-agent") source.Title = "Cross-agent branch" @@ -85,6 +104,44 @@ func TestForkSessionEndpointCanTargetAnotherAgent(t *testing.T) { } } +func TestForkSessionEndpointRejectsUnknownTargetAgent(t *testing.T) { + shannonDir := t.TempDir() + deps := &ServerDeps{ + ShannonDir: shannonDir, + AgentsDir: filepath.Join(shannonDir, "agents"), + SessionCache: NewSessionCache(shannonDir), + } + server := NewServer(0, nil, deps, "test") + mgr := deps.SessionCache.GetOrCreateManager(deps.SessionCache.SessionsDir("")) + source := mgr.NewSessionWithID("source-session-unknown-target") + source.Messages = []client.Message{ + {Role: "user", Content: client.NewTextContent("question")}, + {Role: "assistant", Content: client.NewTextContent("answer")}, + } + if err := mgr.Save(); err != nil { + t.Fatal(err) + } + + body := bytes.NewBufferString(`{"message_index":2,"target_agent":"nope"}`) + req := httptest.NewRequest(http.MethodPost, "/sessions/source-session-unknown-target/fork", body) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + server.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String()) + } + var payload struct { + Code string `json:"code"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &payload); err != nil || payload.Code != "agent_not_found" { + t.Fatalf("code = %q err=%v body=%s", payload.Code, err, rec.Body.String()) + } + // A rejected fork must not litter an orphan session directory. + if _, err := os.Stat(filepath.Join(shannonDir, "agents", "nope")); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("orphan target agent dir created: %v", err) + } +} + func TestForkSessionEndpointRejectsIncompleteTurn(t *testing.T) { shannonDir := t.TempDir() deps := &ServerDeps{ShannonDir: shannonDir, SessionCache: NewSessionCache(shannonDir)} @@ -178,3 +235,111 @@ func TestSideChatEndpointUsesReadOnlyEphemeralContext(t *testing.T) { t.Fatalf("side chat persisted an extra session: %#v", summaries) } } + +// TestSideChatEndpointStreamsSSE covers the transport Desktop actually uses: +// Accept: text/event-stream routes through handleMessageSSE, and the ephemeral +// guarantees (no extra session, no global events) must hold there too. +func TestSideChatEndpointStreamsSSE(t *testing.T) { + gateway := &fakeGatewayBackend{reply: "streamed answer"} + gatewayServer := httptest.NewServer(gateway.handler()) + defer gatewayServer.Close() + + deps := runAgentContractTestDeps(t, gatewayServer.URL) + defer deps.SessionCache.CloseAll() + mgr := deps.SessionCache.GetOrCreateManager(deps.SessionCache.SessionsDir("")) + source := mgr.NewSessionWithID("side-chat-sse-source") + source.Messages = []client.Message{ + {Role: "user", Content: client.NewTextContent("original question")}, + {Role: "assistant", Content: client.NewTextContent("original answer")}, + } + if err := mgr.Save(); err != nil { + t.Fatal(err) + } + + server := NewServer(0, nil, deps, "test") + events := server.EventBus().Subscribe() + body := `{"message_index":2,"text":"explain this"}` + req := httptest.NewRequest(http.MethodPost, "/sessions/side-chat-sse-source/side-chat", bytes.NewBufferString(body)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "text/event-stream") + rec := httptest.NewRecorder() + server.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String()) + } + if ct := rec.Header().Get("Content-Type"); !strings.Contains(ct, "text/event-stream") { + t.Fatalf("content-type = %q", ct) + } + stream := rec.Body.String() + if !strings.Contains(stream, "event: done") || !strings.Contains(stream, "streamed answer") { + t.Fatalf("SSE stream missing done frame: %s", stream) + } + select { + case event := <-events: + t.Fatalf("side chat SSE published global event %q", event.Type) + default: + } + summaries, err := mgr.List() + if err != nil { + t.Fatal(err) + } + if len(summaries) != 1 || summaries[0].ID != source.ID { + t.Fatalf("side chat SSE persisted an extra session: %#v", summaries) + } +} + +// TestSideChatUsesCompactionCheckpointHistory pins that a compacted source +// session feeds the side chat its checkpoint+tail model view, not the full +// raw archive the checkpoint already replaced. +func TestSideChatUsesCompactionCheckpointHistory(t *testing.T) { + gateway := &fakeGatewayBackend{reply: "focused answer"} + gatewayServer := httptest.NewServer(gateway.handler()) + defer gatewayServer.Close() + + deps := runAgentContractTestDeps(t, gatewayServer.URL) + defer deps.SessionCache.CloseAll() + mgr := deps.SessionCache.GetOrCreateManager(deps.SessionCache.SessionsDir("")) + source := mgr.NewSessionWithID("side-chat-compacted-source") + source.Messages = []client.Message{ + {Role: "user", Content: client.NewTextContent("archived question")}, + {Role: "assistant", Content: client.NewTextContent("archived bulky answer")}, + {Role: "user", Content: client.NewTextContent("tail question")}, + {Role: "assistant", Content: client.NewTextContent("tail answer")}, + } + source.CompactionCheckpoint = &session.CompactionCheckpoint{ + SchemaVersion: session.CompactionCheckpointSchemaVersion, + ArchiveThroughIndex: 2, + Messages: []client.Message{ + {Role: "user", Content: client.NewTextContent("archived question")}, + {Role: "user", Content: client.NewTextContent(ctxwin.CompactionSummaryPrefix + "summary of the archived turn")}, + }, + } + if err := mgr.Save(); err != nil { + t.Fatal(err) + } + + server := NewServer(0, nil, deps, "test") + body := `{"message_index":4,"text":"explain this"}` + req := httptest.NewRequest(http.MethodPost, "/sessions/side-chat-compacted-source/side-chat", bytes.NewBufferString(body)) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + server.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String()) + } + + var transcript strings.Builder + for _, request := range gateway.requests() { + for _, message := range request.Messages { + transcript.WriteString(message.Content.Text()) + transcript.WriteByte('\n') + } + } + if !strings.Contains(transcript.String(), "summary of the archived turn") || + !strings.Contains(transcript.String(), "tail question") { + t.Fatalf("side chat lost checkpoint+tail view: %s", transcript.String()) + } + if strings.Contains(transcript.String(), "archived bulky answer") { + t.Fatalf("side chat re-fed the raw archive past the checkpoint: %s", transcript.String()) + } +} diff --git a/internal/daemon/conversation_reply.go b/internal/daemon/conversation_reply.go index 414e331b..d89b98b0 100644 --- a/internal/daemon/conversation_reply.go +++ b/internal/daemon/conversation_reply.go @@ -2,8 +2,12 @@ package daemon import ( "encoding/xml" + "errors" + "fmt" + "io" "strings" + "github.com/Kocoro-lab/ShanClaw/internal/agent" "github.com/Kocoro-lab/ShanClaw/internal/client" "github.com/Kocoro-lab/ShanClaw/internal/session" ) @@ -16,6 +20,14 @@ const ( maxConversationAnnotationCommentRunes = 2_000 ) +// Stable {"error","code"} identifiers for reply-envelope ingress validation. +const ( + conversationRepliesMalformedCode = "conversation_replies_malformed" + conversationRepliesTooManyCode = "conversation_replies_too_many" + conversationReplyQuoteTooLongCode = "conversation_reply_quote_too_long" + conversationReplyCommentTooLongCode = "conversation_reply_comment_too_long" +) + type conversationReplyEnvelope struct { Replies []struct { Quote string `xml:"quote"` @@ -24,17 +36,33 @@ type conversationReplyEnvelope struct { } // splitConversationReplyPrompt separates Desktop's model-only reply envelope -// from the user-visible prompt. The envelope is accepted only at the head of a +// from the user-visible prompt. Envelopes are accepted only at the head of a // message; ordinary user text containing the same tags remains ordinary text. +// Consecutive head envelopes (produced when a drained mailbox batch merges +// with the triggering message) are consumed as one envelope run. func splitConversationReplyPrompt(text string) (visible string, envelope string) { - if !strings.HasPrefix(text, conversationRepliesOpen) { - return text, "" + end := 0 + rest := text + for strings.HasPrefix(rest, conversationRepliesOpen) { + closeAt := strings.Index(rest, conversationRepliesClose) + if closeAt < 0 { + break + } + consumed := closeAt + len(conversationRepliesClose) + end += consumed + rest = rest[consumed:] + // Only step over whitespace when another envelope follows — trailing + // whitespace before visible text belongs to the visible split below. + trimmed := strings.TrimLeft(rest, " \t\r\n") + if !strings.HasPrefix(trimmed, conversationRepliesOpen) { + break + } + end += len(rest) - len(trimmed) + rest = trimmed } - end := strings.Index(text, conversationRepliesClose) - if end < 0 { + if end == 0 { return text, "" } - end += len(conversationRepliesClose) return strings.TrimSpace(text[end:]), text[:end] } @@ -48,20 +76,21 @@ func restoreConversationReplyPrompt(visible, envelope string) string { return envelope + "\n\n" + visible } +// conversationReplyAnnotations decodes one or more concatenated envelopes. +// A parse failure returns nil; a successful parse always returns a non-nil +// slice (possibly empty) so callers can distinguish "malformed" from +// "well-formed but content-free". func conversationReplyAnnotations(envelope string) []session.ConversationAnnotation { - if envelope == "" { - return nil - } - var parsed conversationReplyEnvelope - if err := xml.Unmarshal([]byte(envelope), &parsed); err != nil { + parsed, err := decodeConversationReplyEnvelopes(envelope) + if err != nil || parsed == nil { return nil } - limit := len(parsed.Replies) + limit := len(parsed) if limit > maxConversationAnnotations { limit = maxConversationAnnotations } annotations := make([]session.ConversationAnnotation, 0, limit) - for _, reply := range parsed.Replies[:limit] { + for _, reply := range parsed[:limit] { quote := boundedConversationAnnotationText(reply.Quote, maxConversationAnnotationQuoteRunes) comment := boundedConversationAnnotationText(reply.Comment, maxConversationAnnotationCommentRunes) if quote == "" && comment == "" { @@ -75,53 +104,165 @@ func conversationReplyAnnotations(envelope string) []session.ConversationAnnotat return annotations } -func boundedConversationAnnotationText(text string, maxRunes int) string { - runes := []rune(text) - if len(runes) <= maxRunes { - return text - } - return string(runes[:maxRunes]) +type conversationReplyEntry struct { + Quote string + Comment string } -// visibleConversationReplyText removes every well-formed model-only envelope -// from a combined prompt. Multiple envelopes occur when injected follow-ups -// are drained into one run. Persisted transcript, titles, previews, and global -// events must contain only the user's visible text. -func visibleConversationReplyText(text string) string { - removed := false +// decodeConversationReplyEnvelopes parses a run of concatenated envelopes into +// the raw (unbounded) reply list. ("", nil, nil) for empty input; a non-nil +// error for any parse failure. +func decodeConversationReplyEnvelopes(envelope string) ([]conversationReplyEntry, error) { + if envelope == "" { + return nil, nil + } + dec := xml.NewDecoder(strings.NewReader(envelope)) + replies := []conversationReplyEntry{} for { - start := strings.Index(text, conversationRepliesOpen) - if start < 0 { - if !removed { - return text - } - return strings.TrimSpace(text) + var parsed conversationReplyEnvelope + err := dec.Decode(&parsed) + if errors.Is(err, io.EOF) { + return replies, nil + } + if err != nil { + return nil, err } - relativeEnd := strings.Index(text[start:], conversationRepliesClose) - if relativeEnd < 0 { - return strings.TrimSpace(text) + for _, r := range parsed.Replies { + replies = append(replies, conversationReplyEntry{Quote: r.Quote, Comment: r.Comment}) } - end := start + relativeEnd + len(conversationRepliesClose) - text = text[:start] + text[end:] - removed = true } } -func visibleConversationReplyMessage(message client.Message) client.Message { +// validateConversationReplyEnvelope enforces the reply wire limits on the +// exact envelope the model would receive — not the truncated persistence +// projection. Returns ("", "") for text without a head envelope, else a +// stable code + English fallback message for writeErrorCode. +func validateConversationReplyEnvelope(text string) (code, message string) { + _, envelope := splitConversationReplyPrompt(text) + if envelope == "" { + return "", "" + } + replies, err := decodeConversationReplyEnvelopes(envelope) + if err != nil { + return conversationRepliesMalformedCode, "reply envelope is not well-formed" + } + if len(replies) > maxConversationAnnotations { + return conversationRepliesTooManyCode, + fmt.Sprintf("reply envelope exceeds %d replies", maxConversationAnnotations) + } + for _, reply := range replies { + if len([]rune(reply.Quote)) > maxConversationAnnotationQuoteRunes { + return conversationReplyQuoteTooLongCode, + fmt.Sprintf("reply quote exceeds %d characters", maxConversationAnnotationQuoteRunes) + } + if len([]rune(reply.Comment)) > maxConversationAnnotationCommentRunes { + return conversationReplyCommentTooLongCode, + fmt.Sprintf("reply comment exceeds %d characters", maxConversationAnnotationCommentRunes) + } + } + return "", "" +} + +func boundedConversationAnnotationText(text string, maxRunes int) string { + runes := []rune(text) + if len(runes) <= maxRunes { + return text + } + return string(runes[:maxRunes]) +} + +// conversationReplyPersistedMessage returns the archival copy of a run user +// message plus the annotations decoded from its head envelope. Stripping is +// gated on a successful parse: a delimited-but-malformed envelope is kept +// verbatim so the user's bytes are never silently dropped, and tags anywhere +// past the head are ordinary text. +func conversationReplyPersistedMessage(message client.Message) (client.Message, []session.ConversationAnnotation) { if message.Role != "user" { - return message + return message, nil } if !message.Content.HasBlocks() { - message.Content = client.NewTextContent(visibleConversationReplyText(message.Content.Text())) - return message + visible, annotations, stripped := conversationReplyVisibleText(message.Content.Text()) + if stripped { + message.Content = client.NewTextContent(visible) + } + return message, annotations } blocks := append([]client.ContentBlock(nil), message.Content.Blocks()...) + var annotations []session.ConversationAnnotation for index := range blocks { if blocks[index].Type == "text" { - blocks[index].Text = visibleConversationReplyText(blocks[index].Text) + visible, decoded, stripped := conversationReplyVisibleText(blocks[index].Text) + if stripped { + blocks[index].Text = visible + } + annotations = decoded break } } message.Content = client.NewBlockContent(blocks) - return message + return message, annotations +} + +// conversationReplyVisibleText splits + decodes a head envelope. stripped is +// false (and visible == text) when there is no envelope or it fails to parse. +// An injected-turn message (the agent loop's "[New message from user]" join of +// drained follow-ups) is handled segment-aware, because each follow-up's +// envelope sits at that follow-up's head inside the joined text, not at the +// message head. +func conversationReplyVisibleText(text string) (visible string, annotations []session.ConversationAnnotation, stripped bool) { + if rest, ok := strings.CutPrefix(text, agent.InjectedUserMessagePrefix); ok { + visible, annotations, stripped = stripJoinedConversationReplies(rest) + return agent.InjectedUserMessagePrefix + visible, annotations, stripped + } + split, envelope := splitConversationReplyPrompt(text) + if envelope == "" { + return text, nil, false + } + annotations = conversationReplyAnnotations(envelope) + if annotations == nil { + return text, nil, false + } + return split, annotations, true +} + +// stripJoinedConversationReplies removes every well-formed envelope run that +// starts at the head of a "\n\n"-joined segment — the positions where a +// drained follow-up's own head can land inside an injected-turn message. +// Anything that fails to parse, or sits anywhere else, stays verbatim. +func stripJoinedConversationReplies(text string) (string, []session.ConversationAnnotation, bool) { + var out strings.Builder + var annotations []session.ConversationAnnotation + stripped := false + i := 0 + for { + idx := strings.Index(text[i:], conversationRepliesOpen) + if idx < 0 { + out.WriteString(text[i:]) + break + } + abs := i + idx + atSegmentHead := abs == 0 || strings.HasSuffix(text[:abs], "\n\n") + if atSegmentHead { + segVisible, envelope := splitConversationReplyPrompt(text[abs:]) + if decoded := conversationReplyAnnotations(envelope); envelope != "" && decoded != nil { + out.WriteString(text[:abs][i:]) + annotations = append(annotations, decoded...) + if len(annotations) > maxConversationAnnotations { + annotations = annotations[:maxConversationAnnotations] + } + stripped = true + // splitConversationReplyPrompt already trimmed the separator + // between the envelope run and the segment's visible text. + text = segVisible + i = 0 + continue + } + } + out.WriteString(text[i : abs+len(conversationRepliesOpen)]) + i = abs + len(conversationRepliesOpen) + } + if !stripped { + return text, nil, false + } + return out.String(), annotations, true } diff --git a/internal/daemon/conversation_reply_test.go b/internal/daemon/conversation_reply_test.go index b5dbbf3c..c16aaf19 100644 --- a/internal/daemon/conversation_reply_test.go +++ b/internal/daemon/conversation_reply_test.go @@ -2,10 +2,16 @@ package daemon import ( "context" + "encoding/json" + "net/http" "net/http/httptest" "strings" + "sync" "testing" + "time" + "github.com/Kocoro-lab/ShanClaw/internal/agent" + "github.com/Kocoro-lab/ShanClaw/internal/agenttypes" "github.com/Kocoro-lab/ShanClaw/internal/client" "github.com/Kocoro-lab/ShanClaw/internal/session" ) @@ -33,19 +39,52 @@ func TestConversationReplyAnnotationsDecodeEscapingAndBounds(t *testing.T) { } } -func TestVisibleConversationReplyTextCleansInjectedMessages(t *testing.T) { - raw := "queued\nsecret context\nnext" - if got := visibleConversationReplyText(raw); got != "queued\n\nnext" { - t.Fatalf("visible = %q", got) - } +func TestConversationReplyPersistedMessageStripsHeadEnvelopeOnly(t *testing.T) { + head := "contextnote\n\nvisible" message := client.Message{Role: "user", Content: client.NewBlockContent([]client.ContentBlock{ - {Type: "text", Text: raw}, + {Type: "text", Text: head}, {Type: "image", Source: &client.ImageSource{Type: "base64", MediaType: "image/png", Data: "abc"}}, })} - clean := visibleConversationReplyMessage(message) - if clean.Content.Blocks()[0].Text != "queued\n\nnext" || clean.Content.Blocks()[1].Type != "image" { + clean, annotations := conversationReplyPersistedMessage(message) + if clean.Content.Blocks()[0].Text != "visible" || clean.Content.Blocks()[1].Type != "image" { t.Fatalf("cleaned blocks = %#v", clean.Content.Blocks()) } + if len(annotations) != 1 || annotations[0].SelectedText != "context" { + t.Fatalf("annotations = %#v", annotations) + } + + // A well-formed pair PAST the head is the user's own text — e.g. pasted + // code discussing this feature — and must persist byte-for-byte. + midText := "queued\nq\nnext" + kept, midAnnotations := conversationReplyPersistedMessage(client.Message{ + Role: "user", Content: client.NewTextContent(midText), + }) + if kept.Content.Text() != midText || midAnnotations != nil { + t.Fatalf("mid-text envelope was rewritten: %q %#v", kept.Content.Text(), midAnnotations) + } +} + +func TestConversationReplyPersistedMessageKeepsMalformedEnvelopeVerbatim(t *testing.T) { + malformed := "broken\n\nquery" + kept, annotations := conversationReplyPersistedMessage(client.Message{ + Role: "user", Content: client.NewTextContent(malformed), + }) + if kept.Content.Text() != malformed || annotations != nil { + t.Fatalf("malformed envelope was dropped: %q %#v", kept.Content.Text(), annotations) + } +} + +func TestSplitConversationReplyPromptConsumesConsecutiveEnvelopes(t *testing.T) { + first := "aone" + second := "btwo" + visible, envelope := splitConversationReplyPrompt(first + "\n" + second + "\n\ntail") + if visible != "tail" || envelope != first+"\n"+second { + t.Fatalf("split = %q / %q", visible, envelope) + } + got := conversationReplyAnnotations(envelope) + if len(got) != 2 || got[0].Comment != "one" || got[1].Comment != "two" { + t.Fatalf("merged annotations = %#v", got) + } } func TestConversationReplyTagsInsideOrdinaryTextAreNotSplit(t *testing.T) { @@ -54,8 +93,32 @@ func TestConversationReplyTagsInsideOrdinaryTextAreNotSplit(t *testing.T) { if visible != raw || envelope != "" { t.Fatalf("ordinary text changed: visible=%q envelope=%q", visible, envelope) } - if got := visibleConversationReplyText(" ordinary text "); got != " ordinary text " { - t.Fatalf("ordinary visible text changed: %q", got) +} + +func TestValidateConversationReplyEnvelope(t *testing.T) { + long := strings.Repeat("字", maxConversationAnnotationQuoteRunes+1) + longComment := strings.Repeat("字", maxConversationAnnotationCommentRunes+1) + var many strings.Builder + many.WriteString(conversationRepliesOpen) + for range maxConversationAnnotations + 1 { + many.WriteString("qc") + } + many.WriteString(conversationRepliesClose) + + cases := []struct { + name, text, code string + }{ + {"no envelope", "plain text", ""}, + {"valid", "qc\n\nok", ""}, + {"malformed", "x", conversationRepliesMalformedCode}, + {"too many", many.String(), conversationRepliesTooManyCode}, + {"quote too long", "" + long + "", conversationReplyQuoteTooLongCode}, + {"comment too long", "" + longComment + "", conversationReplyCommentTooLongCode}, + } + for _, tc := range cases { + if code, _ := validateConversationReplyEnvelope(tc.text); code != tc.code { + t.Fatalf("%s: code = %q, want %q", tc.name, code, tc.code) + } } } @@ -110,3 +173,157 @@ func TestConversationReplyContextReachesModelButNotPersistedSession(t *testing.T } } } + +func TestConversationReplyIngressValidationRejectsOversizedEnvelope(t *testing.T) { + shannonDir := t.TempDir() + deps := &ServerDeps{ShannonDir: shannonDir, SessionCache: NewSessionCache(shannonDir)} + server := NewServer(0, nil, deps, "test") + + malformed := `{"text":"x"}` + for _, target := range []string{"/message", "/queue"} { + body := malformed + if target == "/queue" { + body = `{"route_key":"session:reply-validate","text":"x"}` + } + req := httptest.NewRequest(http.MethodPost, target, strings.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + server.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusBadRequest { + t.Fatalf("POST %s = %d body=%s", target, rec.Code, rec.Body.String()) + } + var payload struct { + Code string `json:"code"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &payload); err != nil || payload.Code != conversationRepliesMalformedCode { + t.Fatalf("POST %s code = %q err=%v", target, payload.Code, err) + } + } +} + +// TestConversationReplyDrainedMailboxAnnotationsPersist pins the busy-queue +// path: an annotated message enqueued while the route was idle drains into the +// next run's merged user turn, and its annotations must survive persistence +// exactly like the trigger message's do (they previously vanished on reload). +func TestConversationReplyDrainedMailboxAnnotationsPersist(t *testing.T) { + const sessID = "drained-annotations-persist" + routeKey := "session:" + sanitizeRouteValue(sessID) + + deps, _ := mailboxOrderingDeps(t) + defer deps.SessionCache.CloseAll() + gateway := &fakeGatewayBackend{reply: "ack"} + ts := httptest.NewServer(gateway.handler()) + defer ts.Close() + deps.GW = client.NewGatewayClient(ts.URL, "test-key") + + queued := agenttypes.QueuedMessage{ + ID: "mbx-annotated-1", + Source: "desktop", + Text: "queued quotequeued comment\n\nqueued visible", + Priority: agenttypes.PriorityNext, + EnqueuedAt: time.Now(), + } + if out, err := deps.SessionCache.EnqueueMessage(routeKey, queued); err != nil || out != MailboxQueued { + t.Fatalf("seed enqueue: out=%v err=%v", out, err) + } + + trigger := "live quotelive comment\n\nlive visible" + if _, err := RunAgent(context.Background(), deps, RunAgentRequest{ + Text: trigger, + SessionID: sessID, + NewSession: true, + Source: "desktop", + }, nullEventHandler{}); err != nil { + t.Fatalf("RunAgent: %v", err) + } + + var modelInput strings.Builder + for _, request := range gateway.requests() { + for _, message := range request.Messages { + modelInput.WriteString(message.Content.Text()) + } + } + for _, want := range []string{"queued quote", "live quote", "queued visible", "live visible"} { + if !strings.Contains(modelInput.String(), want) { + t.Fatalf("model input omitted %q: %s", want, modelInput.String()) + } + } + + mgr := deps.SessionCache.GetOrCreateManager(deps.SessionCache.SessionsDir("")) + persisted, err := mgr.Load(sessID) + if err != nil { + t.Fatal(err) + } + if len(persisted.Messages) == 0 || len(persisted.MessageMeta) == 0 { + t.Fatalf("nothing persisted: %#v", persisted.Messages) + } + if got := persisted.Messages[0].Content.Text(); got != "queued visible\nlive visible" { + t.Fatalf("merged visible text = %q", got) + } + annotations := persisted.MessageMeta[0].ConversationAnnotations + if len(annotations) != 2 || + annotations[0] != (session.ConversationAnnotation{SelectedText: "queued quote", Comment: "queued comment"}) || + annotations[1] != (session.ConversationAnnotation{SelectedText: "live quote", Comment: "live comment"}) { + t.Fatalf("merged annotations = %#v", annotations) + } + for _, message := range persisted.Messages { + if strings.Contains(message.Content.Text(), conversationRepliesOpen) { + t.Fatalf("persisted message leaked envelope: %q", message.Content.Text()) + } + } +} + +// TestApplyTurnMessagesPersistsMidRunInjectedAnnotations pins the other half +// of the busy-queue path: a follow-up injected into a RUNNING loop becomes its +// own user message, whose head envelope must decode into persisted annotations +// instead of evaporating. +func TestApplyTurnMessagesPersistsMidRunInjectedAnnotations(t *testing.T) { + injectCh := make(chan agent.InjectedMessage, 1) + var calls int + var mu sync.Mutex + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + calls++ + n := calls + mu.Unlock() + if n == 1 { + injectCh <- agent.InjectedMessage{ + Text: "mid quotemid comment\n\nmid visible", + } + } + _ = json.NewEncoder(w).Encode(client.CompletionResponse{ + Model: "test-model", OutputText: "done", FinishReason: "end_turn", + Usage: client.Usage{InputTokens: 1, OutputTokens: 1, TotalTokens: 2}, RequestID: "r", + }) + })) + defer ts.Close() + + gw := client.NewGatewayClient(ts.URL, "") + loop := agent.NewAgentLoop(gw, agent.NewToolRegistry(), "medium", "", 10, 2000, 200, nil, nil, nil) + loop.SetInjectCh(injectCh) + if _, _, err := loop.Run(context.Background(), "hello", nil, nil); err != nil { + t.Fatalf("Run: %v", err) + } + + sess := &session.Session{ID: "mid-run-annotations"} + b := captureTurnBaseline(sess, "desktop", false) + applyTurnMessages(sess, loop, b) + + var injectedIdx = -1 + for i, msg := range sess.Messages { + if msg.Role == "user" && strings.Contains(msg.Content.Text(), "mid visible") { + injectedIdx = i + } + if strings.Contains(msg.Content.Text(), conversationRepliesOpen) { + t.Fatalf("persisted message %d leaked envelope: %q", i, msg.Content.Text()) + } + } + if injectedIdx < 0 { + t.Fatalf("injected message not persisted: %#v", sess.Messages) + } + annotations := sess.MessageMeta[injectedIdx].ConversationAnnotations + if len(annotations) != 1 || + annotations[0] != (session.ConversationAnnotation{SelectedText: "mid quote", Comment: "mid comment"}) { + t.Fatalf("injected annotations = %#v", annotations) + } +} diff --git a/internal/daemon/queue_enqueue_handler.go b/internal/daemon/queue_enqueue_handler.go index 0c162e18..18bbd208 100644 --- a/internal/daemon/queue_enqueue_handler.go +++ b/internal/daemon/queue_enqueue_handler.go @@ -83,6 +83,12 @@ func (s *Server) handleEnqueueQueue(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusRequestEntityTooLarge, "message text exceeds 1 MB cap") return } + // Queued messages drain into a run's prompt, so the reply-envelope limits + // apply at this ingress exactly as they do on POST /message. + if code, msg := validateConversationReplyEnvelope(req.Text); code != "" { + writeErrorCode(w, http.StatusBadRequest, code, msg) + return + } editable := true if req.Editable != nil { diff --git a/internal/daemon/runner.go b/internal/daemon/runner.go index f96fba30..0214f951 100644 --- a/internal/daemon/runner.go +++ b/internal/daemon/runner.go @@ -2585,6 +2585,13 @@ func RunAgent(ctx context.Context, deps *ServerDeps, req RunAgentRequest, handle agentName := req.Agent visibleInput, replyEnvelope := splitConversationReplyPrompt(req.Text) replyAnnotations := conversationReplyAnnotations(replyEnvelope) + if replyEnvelope != "" && replyAnnotations == nil { + // Delimited but unparseable: treat the whole message as ordinary text + // so the user's bytes survive verbatim. Desktop-facing ingresses 400 + // this shape up front (validateConversationReplyEnvelope); this covers + // every path that cannot reject. + visibleInput, replyEnvelope = req.Text, "" + } prompt := visibleInput // Download remote file attachments and convert to file_ref blocks. @@ -2818,11 +2825,16 @@ func RunAgent(ctx context.Context, deps *ServerDeps, req RunAgentRequest, handle // consumed and recovery re-delivers them on every restart. if pendingBatch := deps.SessionCache.DrainMailbox(req.RouteKey, 20); len(pendingBatch) > 0 { pendingIDs := make([]string, 0, len(pendingBatch)) - var b strings.Builder + // Each drained message is split individually: its head envelope + // (raw Desktop bytes) joins the merged envelope run, its decoded + // annotations join the merged turn's metadata, and only its + // visible text enters the merged visible prompt. A malformed + // envelope stays verbatim in the visible text (lossless). + var visibleParts []string + var envelopeParts []string + var drainedAnnotations []session.ConversationAnnotation for _, m := range pendingBatch { - if b.Len() > 0 { - b.WriteByte('\n') - } + var b strings.Builder b.WriteString(m.Text) // Phase 4: surface any attachments as a bracketed hint so the // LLM is aware the user shipped files alongside this text. @@ -2851,15 +2863,38 @@ func RunAgent(ctx context.Context, deps *ServerDeps, req RunAgentRequest, handle } b.WriteByte(']') } + visible, annotations, stripped := conversationReplyVisibleText(b.String()) + if stripped { + _, envelope := splitConversationReplyPrompt(b.String()) + envelopeParts = append(envelopeParts, envelope) + drainedAnnotations = append(drainedAnnotations, annotations...) + } + if visible != "" { + visibleParts = append(visibleParts, visible) + } pendingIDs = append(pendingIDs, m.ID) } - if b.Len() > 0 { - if prompt == "" { - prompt = b.String() - } else { - prompt = b.String() + "\n" + prompt + if len(visibleParts) > 0 || len(envelopeParts) > 0 { + mergedVisible := strings.Join(visibleParts, "\n") + if visiblePrompt != "" { + if mergedVisible != "" { + mergedVisible += "\n" + } + mergedVisible += visiblePrompt } - visiblePrompt = visibleConversationReplyText(prompt) + // Drained messages precede the trigger in the merged text, so + // their envelopes and annotations lead in the same order. + if replyEnvelope != "" { + envelopeParts = append(envelopeParts, replyEnvelope) + } + replyEnvelope = strings.Join(envelopeParts, "\n") + merged := append(drainedAnnotations, replyAnnotations...) + if len(merged) > maxConversationAnnotations { + merged = merged[:maxConversationAnnotations] + } + replyAnnotations = merged + visiblePrompt = mergedVisible + prompt = restoreConversationReplyPrompt(visiblePrompt, replyEnvelope) req.Text = visiblePrompt } log.Printf("daemon: drained %d mailbox msg(s) into prompt for route %q", len(pendingBatch), req.RouteKey) @@ -5256,20 +5291,38 @@ func applyTurnMessages(sess *session.Session, loop *agent.AgentLoop, b turnBasel if b.preLoopUser && runMsgs[0].Role == "user" { startIdx = 1 } + // The triggering user message's annotations were decoded at ingress (they + // may merge a drained mailbox batch); reuse them instead of re-decoding. + // When preLoopUser is set the trigger was persisted before the loop with + // its annotations attached, so no run message may claim them here. + triggerIdx := -1 + if !b.preLoopUser { + triggerIdx = startIdx + } fallbackTime := time.Now() for i := startIdx; i < len(runMsgs); i++ { ts := fallbackTime if i < len(runTimestamps) && !runTimestamps[i].IsZero() { ts = runTimestamps[i] } - sess.Messages = append(sess.Messages, visibleConversationReplyMessage(runMsgs[i])) + injected := i < len(runInjected) && runInjected[i] + persisted := runMsgs[i] meta := session.MessageMeta{Source: b.source, Timestamp: session.TimePtr(ts)} - if i == startIdx && runMsgs[i].Role == "user" { - meta.ConversationAnnotations = b.conversationAnnotations - } - if i < len(runInjected) && runInjected[i] { + if injected { meta.SystemInjected = true } + if persisted.Role == "user" && !injected { + // Mid-run injected follow-ups carry their own head envelopes: + // decode each one so its annotations survive reload, exactly like + // the trigger message's do. + var annotations []session.ConversationAnnotation + persisted, annotations = conversationReplyPersistedMessage(persisted) + if i == triggerIdx { + annotations = b.conversationAnnotations + } + meta.ConversationAnnotations = annotations + } + sess.Messages = append(sess.Messages, persisted) sess.MessageMeta = append(sess.MessageMeta, meta) } } diff --git a/internal/daemon/server.go b/internal/daemon/server.go index 9ad519e5..79ee530e 100644 --- a/internal/daemon/server.go +++ b/internal/daemon/server.go @@ -3179,6 +3179,12 @@ func (s *Server) handleMessage(w http.ResponseWriter, r *http.Request) { return } } + // Reply-envelope limits are enforced on the exact bytes the model would + // receive, not on the truncated persistence projection. + if code, msg := validateConversationReplyEnvelope(req.Text); code != "" { + writeErrorCode(w, http.StatusBadRequest, code, msg) + return + } if req.Source == "" { req.Source = "kocoro" } diff --git a/internal/daemon/wire_fixtures_test.go b/internal/daemon/wire_fixtures_test.go index 8a8b4a02..96f8b6d5 100644 --- a/internal/daemon/wire_fixtures_test.go +++ b/internal/daemon/wire_fixtures_test.go @@ -2397,3 +2397,48 @@ func TestWireFixture_DoneWithExecutionRun(t *testing.T) { t.Fatalf("consumer decode lost execution_run fields: %+v", done.ExecutionRun) } } + +// TestWireFixture_HTTPSessionFork pins the fork response shape through the +// real POST /sessions/{id}/fork route. The fork's session id is generated, so +// it is shape-asserted (non-empty date-hex id) and normalized to the fixture. +func TestWireFixture_HTTPSessionFork(t *testing.T) { + fixture := loadWireFixture(t, "http_post.session_fork.response.json") + + shannonDir := t.TempDir() + deps := &ServerDeps{ShannonDir: shannonDir, SessionCache: NewSessionCache(shannonDir)} + server := NewServer(0, nil, deps, "test") + mgr := deps.SessionCache.GetOrCreateManager(deps.SessionCache.SessionsDir("")) + source := mgr.NewSessionWithID("wire-fixture-fork-source") + source.Title = fixture["title"].(string) + source.Messages = []client.Message{ + {Role: "user", Content: client.NewTextContent("question")}, + {Role: "assistant", Content: client.NewTextContent("answer")}, + } + if err := mgr.Save(); err != nil { + t.Fatal(err) + } + + req := httptest.NewRequest(http.MethodPost, "/sessions/wire-fixture-fork-source/fork", + strings.NewReader(`{"message_index":2}`)) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + server.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusCreated { + t.Fatalf("POST /sessions/{id}/fork = %d: %s", rec.Code, rec.Body.Bytes()) + } + produced := parseJSONMap(t, rec.Body.Bytes()) + if v, ok := produced["session_id"].(string); !ok || v == "" || v == "wire-fixture-fork-source" { + t.Fatalf("session_id: want fresh generated id, got %#v", produced["session_id"]) + } + produced["session_id"] = fixture["session_id"] + assertSemanticEqual(t, fixture, produced) + + // Decode the produced bytes through the same consumer shape Desktop uses. + var response forkSessionResponse + if err := json.Unmarshal(rec.Body.Bytes(), &response); err != nil { + t.Fatalf("consumer decode failed: %v", err) + } + if response.Title != fixture["title"].(string) { + t.Fatalf("consumer decode lost title: %+v", response) + } +} diff --git a/internal/session/fork_test.go b/internal/session/fork_test.go index 47f234ee..1a21cb77 100644 --- a/internal/session/fork_test.go +++ b/internal/session/fork_test.go @@ -8,6 +8,7 @@ import ( "time" "github.com/Kocoro-lab/ShanClaw/internal/client" + ctxwin "github.com/Kocoro-lab/ShanClaw/internal/context" ) func TestForkSessionCopiesConversationWithoutRuntimeState(t *testing.T) { @@ -149,3 +150,138 @@ func TestCopyHistoryThroughDropsInjectedMessagesAndDoesNotAlias(t *testing.T) { t.Fatal("copied history aliases the persisted source") } } + +// compactedSourceSession builds a session whose first two turns are covered by +// a valid compaction checkpoint (marker included) and whose tail continues past +// it, mirroring what daemon auto-compaction persists. +func compactedSourceSession(mgr *Manager, id string) *Session { + source := mgr.NewSessionWithID(id) + source.Messages = []client.Message{ + mkMsg("user", "first question"), + mkMsg("assistant", "first answer"), + mkMsg("user", "second question"), + mkMsg("assistant", "second answer"), + mkMsg("user", "tail question"), + mkMsg("assistant", "tail answer"), + } + source.MessageMeta = make([]MessageMeta, len(source.Messages)) + source.CompactionCheckpoint = &CompactionCheckpoint{ + SchemaVersion: CompactionCheckpointSchemaVersion, + ArchiveThroughIndex: 4, + Messages: []client.Message{ + mkMsg("user", "first question"), + mkMsg("user", ctxwin.CompactionSummaryPrefix+"compacted summary of the first turns"), + }, + } + return source +} + +func TestForkSessionCarriesCoveringCompactionCheckpoint(t *testing.T) { + mgr := NewManager(t.TempDir()) + source := compactedSourceSession(mgr, "compacted-fork-source") + if err := mgr.Save(); err != nil { + t.Fatal(err) + } + + fork, err := mgr.ForkSession(source.ID, 6) + if err != nil { + t.Fatal(err) + } + if fork.CompactionCheckpoint == nil { + t.Fatal("fork dropped the covering compaction checkpoint") + } + history := fork.HistoryForLoop() + if len(history) != 4 { + t.Fatalf("fork history = %d messages, want checkpoint(2)+tail(2)", len(history)) + } + if history[1].Content.Text() != ctxwin.CompactionSummaryPrefix+"compacted summary of the first turns" || + history[2].Content.Text() != "tail question" { + t.Fatalf("fork history is not checkpoint+tail: %#v", history) + } + // Deep copy: mutating the fork's checkpoint must not reach the source. + fork.CompactionCheckpoint.Messages[1] = mkMsg("user", "mutated") + reloaded, err := mgr.Load(source.ID) + if err != nil { + t.Fatal(err) + } + if reloaded.CompactionCheckpoint.Messages[1].Content.Text() == "mutated" { + t.Fatal("fork checkpoint aliases the source checkpoint") + } + + // A cut BEFORE the checkpoint's coverage cannot use it — the checkpoint + // summarizes messages the fork does not contain. + early, err := mgr.ForkSession(source.ID, 2) + if err != nil { + t.Fatal(err) + } + if early.CompactionCheckpoint != nil { + t.Fatal("fork kept a checkpoint covering messages beyond its cut") + } +} + +func TestCopyHistoryThroughUsesCoveringCompactionCheckpoint(t *testing.T) { + mgr := NewManager(t.TempDir()) + source := compactedSourceSession(mgr, "compacted-copy-source") + if err := mgr.Save(); err != nil { + t.Fatal(err) + } + + history, err := mgr.CopyHistoryThrough(source.ID, 6) + if err != nil { + t.Fatal(err) + } + if len(history) != 4 || history[1].Content.Text() != ctxwin.CompactionSummaryPrefix+"compacted summary of the first turns" { + t.Fatalf("copy is not checkpoint+tail: %#v", history) + } + + // A boundary inside the covered range falls back to the raw prefix. + rawPrefix, err := mgr.CopyHistoryThrough(source.ID, 2) + if err != nil { + t.Fatal(err) + } + if len(rawPrefix) != 2 || rawPrefix[1].Content.Text() != "first answer" { + t.Fatalf("pre-coverage copy = %#v", rawPrefix) + } +} + +func TestForkSessionIntoConcurrentForksDoNotInterleave(t *testing.T) { + dir := t.TempDir() + mgr := NewManager(dir) + source := mgr.NewSessionWithID("concurrent-fork-source") + source.Messages = []client.Message{ + mkMsg("user", "question"), + mkMsg("assistant", "answer"), + } + if err := mgr.Save(); err != nil { + t.Fatal(err) + } + target := NewManager(filepath.Join(dir, "other-agent")) + + type outcome struct { + fork *Session + err error + } + results := make(chan outcome, 2) + go func() { + fork, err := mgr.ForkSessionInto(source.ID, 2, mgr) + results <- outcome{fork, err} + }() + go func() { + fork, err := mgr.ForkSessionInto(source.ID, 2, target) + results <- outcome{fork, err} + }() + seen := map[string]bool{} + for range 2 { + r := <-results + if r.err != nil { + t.Fatal(r.err) + } + if seen[r.fork.ID] { + t.Fatalf("concurrent forks shared id %q", r.fork.ID) + } + seen[r.fork.ID] = true + if len(r.fork.Messages) != 2 || r.fork.Messages[1].Content.Text() != "answer" { + t.Fatalf("concurrent fork corrupted messages: %#v", r.fork.Messages) + } + } +} diff --git a/internal/session/manager.go b/internal/session/manager.go index fd889d15..eab03ea5 100644 --- a/internal/session/manager.go +++ b/internal/session/manager.go @@ -561,7 +561,10 @@ func (m *Manager) TruncateMessages(id string, index int) error { // CopyHistoryThrough returns an independent, model-visible copy of a session's // history through index. index is exclusive and must end at a complete -// assistant turn. Loop-internal injected messages are omitted. +// assistant turn. Loop-internal injected messages are omitted, and a +// compaction checkpoint covering the requested prefix is honored — the copy is +// the compacted summary plus the retained tail, exactly what a run at that +// boundary would see, not the full raw archive. func (m *Manager) CopyHistoryThrough(id string, index int) ([]client.Message, error) { m.mu.Lock() defer m.mu.Unlock() @@ -573,12 +576,7 @@ func (m *Manager) CopyHistoryThrough(id string, index int) ([]client.Message, er if err := validateCompleteTurnBoundary(sess.Messages, index); err != nil { return nil, err } - meta := sess.MessageMeta - if len(meta) > index { - meta = meta[:index] - } - history := FilterInjected(sess.Messages[:index], meta) - return cloneMessages(history) + return cloneMessages(sess.HistoryThrough(index)) } // ForkSession creates an ordinary persisted session containing the source @@ -613,19 +611,37 @@ func (m *Manager) ForkSessionInto(id string, index int, target *Manager) (*Sessi } metaCount := min(index, len(source.MessageMeta)) meta := append([]MessageMeta(nil), source.MessageMeta[:metaCount]...) + // A checkpoint whose coverage ends at or before the cut carries over so the + // fork's first run reuses the existing summary instead of re-feeding (and + // re-compacting) the full raw archive. Coverage past the cut would describe + // messages the fork does not have — drop it and let the fork self-heal. + var checkpoint *CompactionCheckpoint + if cp := source.CompactionCheckpoint; cp != nil && cp.ArchiveThroughIndex <= index { + checkpointMessages, cpErr := cloneMessages(cp.Messages) + if cpErr != nil { + m.mu.Unlock() + return nil, cpErr + } + checkpoint = &CompactionCheckpoint{ + SchemaVersion: cp.SchemaVersion, + ArchiveThroughIndex: cp.ArchiveThroughIndex, + Messages: checkpointMessages, + } + } now := time.Now() fork := &Session{ - SchemaVersion: 1, - ID: generateID(), - CreatedAt: now, - Title: source.Title, - TitleAuto: source.TitleAuto, - TitleTurns: source.TitleTurns, - CWD: source.CWD, - Messages: messages, - MessageMeta: meta, - Source: source.Source, - ProjectID: source.ProjectID, + SchemaVersion: 1, + ID: generateID(), + CreatedAt: now, + Title: source.Title, + TitleAuto: source.TitleAuto, + TitleTurns: source.TitleTurns, + CWD: source.CWD, + Messages: messages, + MessageMeta: meta, + Source: source.Source, + ProjectID: source.ProjectID, + CompactionCheckpoint: checkpoint, } m.mu.Unlock() diff --git a/internal/session/store.go b/internal/session/store.go index 802abd02..91fc741a 100644 --- a/internal/session/store.go +++ b/internal/session/store.go @@ -339,15 +339,40 @@ func (s *Session) HistoryForLoop() []client.Message { if s == nil { return nil } + return s.HistoryThrough(len(s.Messages)) +} + +// HistoryThrough is HistoryForLoop bounded to the raw archive prefix +// [0, index) — the model-visible view a run would see if the archive ended at +// index. A checkpoint whose coverage extends past index cannot describe that +// prefix, so it falls back to the filtered raw slice (same self-healing as an +// invalid checkpoint). Callers pass an index already validated against the +// archive; it is clamped defensively. +func (s *Session) HistoryThrough(index int) []client.Message { + if s == nil { + return nil + } + if index < 0 { + index = 0 + } + if index > len(s.Messages) { + index = len(s.Messages) + } + meta := s.MessageMeta + if len(meta) > index { + meta = meta[:index] + } cp := s.CompactionCheckpoint if cp == nil { - return FilterInjected(s.Messages, s.MessageMeta) + return FilterInjected(s.Messages[:index], meta) } if cp.SchemaVersion != CompactionCheckpointSchemaVersion || len(cp.Messages) == 0 || - cp.ArchiveThroughIndex < 0 || cp.ArchiveThroughIndex > len(s.Messages) { - log.Printf("session: ignoring invalid compaction checkpoint (session=%q schema=%d messages=%d archive_through_index=%d archive_messages=%d); falling back to archive", - s.ID, cp.SchemaVersion, len(cp.Messages), cp.ArchiveThroughIndex, len(s.Messages)) - return FilterInjected(s.Messages, s.MessageMeta) + cp.ArchiveThroughIndex < 0 || cp.ArchiveThroughIndex > index { + if index == len(s.Messages) { + log.Printf("session: ignoring invalid compaction checkpoint (session=%q schema=%d messages=%d archive_through_index=%d archive_messages=%d); falling back to archive", + s.ID, cp.SchemaVersion, len(cp.Messages), cp.ArchiveThroughIndex, len(s.Messages)) + } + return FilterInjected(s.Messages[:index], meta) } // ShapeHistory always writes the compacted-history marker — even a failed // summary keeps the prefixed "(summary unavailable)" line — so a @@ -358,17 +383,17 @@ func (s *Session) HistoryForLoop() []client.Message { if !ctxwin.IsCompactedHistory(cp.Messages) { log.Printf("session: ignoring compaction checkpoint missing the compacted-history marker (session=%q messages=%d archive_through_index=%d); falling back to archive", s.ID, len(cp.Messages), cp.ArchiveThroughIndex) - return FilterInjected(s.Messages, s.MessageMeta) + return FilterInjected(s.Messages[:index], meta) } through := cp.ArchiveThroughIndex - tailMeta := s.MessageMeta + tailMeta := meta if len(tailMeta) > through { tailMeta = tailMeta[through:] } else { tailMeta = nil } - tail := FilterInjected(s.Messages[through:], tailMeta) + tail := FilterInjected(s.Messages[through:index], tailMeta) out := make([]client.Message, 0, len(cp.Messages)+len(tail)) out = append(out, cp.Messages...) out = append(out, tail...) From a6eca17869d639f3a1f1ee154fefaec4bf41c2f1 Mon Sep 17 00:00:00 2001 From: Shenghao Hu Date: Mon, 17 Aug 2026 16:52:47 +0900 Subject: [PATCH 3/4] feat: run side chats with full tools and detect bare tool_call fabrication MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A side chat that needed web_search emitted a literal web_search(...) as text: the run fed history full of tool traffic into a loop whose registry was emptied by DisableTools, nothing told the model it had no tools, and the fabrication detector only knew the "I called" and shapes. Product decision: a side chat should be capability-identical to the primary conversation — only its lifecycle is ephemeral. - Drop DisableTools from POST /sessions/{id}/side-chat: side chats run the normal tool registry and permission engine. Approvals ride the per-request SSE stream exactly like /message (verified end to end by a live-server test: approval frame -> POST /approval allow -> tool executes -> done). The DisableTools runner mechanism itself is unchanged. - Rewrite the side-chat StickyContext: branched-from-primary framing, tools are available, the primary conversation must not be modified. - Gate the SSE handler's QuestionAsker injection on !req.Ephemeral (shouldInjectQuestionAsker): the side-chat panel has no question UI, so an asker would block for the auto-resolution window and then report a decline the user never made. ask_user_question degrades to its clean "can't ask here" result instead. - Teach fabricatedToolCallPattern the bare paired name(args) block. The branch requires paired tags around a call-shaped body so prose or code discussing the tag stays unflagged. Backstop, independent of the side-chat change. - Contract: update the side-chat entry in docs/cloud-contract-surface.md (tools + approvals + no asker; ephemeral semantics unchanged). Verification: go test ./... passes (full suite, exit 0). New coverage: side-chat request carries the registered tool registry, full SSE approval round-trip on the side-chat route, asker gate table test, fabricated positives plus unpaired/prose negatives. --- docs/cloud-contract-surface.md | 15 +- internal/agent/loop.go | 10 +- internal/agent/loop_test.go | 16 ++ internal/daemon/conversation_context.go | 7 +- internal/daemon/conversation_context_test.go | 177 ++++++++++++++++++- internal/daemon/runner.go | 10 ++ internal/daemon/server.go | 2 +- 7 files changed, 223 insertions(+), 14 deletions(-) diff --git a/docs/cloud-contract-surface.md b/docs/cloud-contract-surface.md index 9735be7b..bd1bc3b0 100644 --- a/docs/cloud-contract-surface.md +++ b/docs/cloud-contract-surface.md @@ -169,11 +169,16 @@ Its optional `agent` identifies the source session directory; optional `target_agent` selects the destination agent directory (use `"default"` for Default, and omit it to branch within the source agent). `POST /sessions/{id}/side-chat` runs against that same bounded history plus the -panel's temporary user/assistant history. Side-chat runs are ephemeral, expose -no tools, and publish no global bus events. Their implementation is -`internal/daemon/conversation_context.go`; the Desktop consumer is -`DaemonClient+ConversationContext.swift`. Neither route belongs in the bundled -Kocoro skill references because the model never calls them. +panel's temporary user/assistant history. Side-chat runs carry the normal tool +registry and permission engine — identical capability to the primary +conversation. Tool approvals flow over the per-request SSE stream +(`approval` frames, resolved via `POST /approval`) exactly like `/message`; +`ask_user_question` gets no asker on ephemeral runs (the panel has no question +UI) and degrades to its clean "can't ask here" result. The runs stay ephemeral: +no session is persisted and no global bus events are published. The +implementation is `internal/daemon/conversation_context.go`; the Desktop +consumer is `DaemonClient+ConversationContext.swift`. Neither route belongs in +the bundled Kocoro skill references because the model never calls them. `message_index` on both routes is a boundary in the RAW archive index space — the `messages` array of the session file, system-injected entries included — diff --git a/internal/agent/loop.go b/internal/agent/loop.go index e0818c45..3e1bc118 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -8887,9 +8887,13 @@ func looksLikeUnverifiedActionClaim(text string) bool { // fabricatedToolCallPattern matches text that mimics tool call output format. // Real tool calls go through the tool_calls API array — they never appear as text. -// Matches both old "I called" format (backward compat) and new XML tags. -// XML branch requires exact attribute shape to avoid false-positives on code examples. -var fabricatedToolCallPattern = regexp.MustCompile(`(?s)(?:I called \w+\(.*?\)\.\s*\n\n(?:Result|Error):\s|\n.*?\n.*?\n)`) +// Matches the old "I called" format (backward compat), the XML tags, +// and a bare paired name(args) block — a shape observed +// when the model invents a call syntax instead of using the API (e.g. a run +// whose history shows tool traffic it cannot reproduce). Each branch stays +// strict — exact attribute shape / paired tags around a function-call body — +// to avoid false-positives on code examples discussing these formats. +var fabricatedToolCallPattern = regexp.MustCompile(`(?s)(?:I called \w+\(.*?\)\.\s*\n\n(?:Result|Error):\s|\n.*?\n.*?\n|\s*\w+\(.*?\)\s*)`) // looksLikeFabricatedToolCalls returns true if the model's text output contains // what looks like fabricated tool call results. This is always a hallucination — diff --git a/internal/agent/loop_test.go b/internal/agent/loop_test.go index 43953153..6940620d 100644 --- a/internal/agent/loop_test.go +++ b/internal/agent/loop_test.go @@ -2698,10 +2698,26 @@ func TestFabricatedToolCallDetection(t *testing.T) { if !looksLikeFabricatedToolCalls(xml) { t.Error("should detect XML format in text output") } + // Bare paired block (observed side-chat fabrication shape) + bare := `Let me search for that. + +web_search(query="kocoro daemon port") + +The results show...` + if !looksLikeFabricatedToolCalls(bare) { + t.Error("should detect paired block") + } // Normal text if looksLikeFabricatedToolCalls("Here is the answer.") { t.Error("should not flag normal text") } + // Unpaired tag and prose mentioning the tag stay unflagged + if looksLikeFabricatedToolCalls("The tag is used by some models.") { + t.Error("should not flag unpaired tag mention") + } + if looksLikeFabricatedToolCalls("Wrap output in plain words markers.") { + t.Error("should not flag paired tags without a call-shaped body") + } } func TestPreambleSuppressedWithToolCalls(t *testing.T) { diff --git a/internal/daemon/conversation_context.go b/internal/daemon/conversation_context.go index e6aba964..64a2d62d 100644 --- a/internal/daemon/conversation_context.go +++ b/internal/daemon/conversation_context.go @@ -130,6 +130,10 @@ func (s *Server) handleSideChat(w http.ResponseWriter, r *http.Request) { } history = append(history, client.Message{Role: role, Content: client.NewTextContent(content)}) } + // Side chats run with the normal tool registry and permission engine — + // identical capability to the primary conversation, only the lifecycle is + // ephemeral. Approvals flow over the per-request SSE stream like any + // /message run. req := RunAgentRequest{ Text: body.Text, Agent: body.Agent, @@ -140,9 +144,8 @@ func (s *Server) handleSideChat(w http.ResponseWriter, r *http.Request) { Ephemeral: true, BypassRouting: true, SessionHistory: history, - DisableTools: true, SuppressBusEvents: true, - StickyContext: "This is a temporary side conversation. Answer from the provided conversation context. Do not modify or claim to modify the primary conversation.", + StickyContext: "This is a temporary side conversation branched from the primary conversation; its transcript is not saved. You have your normal tools — use them when they help answer. Do not modify or claim to modify the primary conversation itself.", } if err := req.Validate(); err != nil { writeError(w, http.StatusBadRequest, err.Error()) diff --git a/internal/daemon/conversation_context_test.go b/internal/daemon/conversation_context_test.go index a6925b57..6d4b475d 100644 --- a/internal/daemon/conversation_context_test.go +++ b/internal/daemon/conversation_context_test.go @@ -1,9 +1,12 @@ package daemon import ( + "bufio" "bytes" + "context" "encoding/json" "errors" + "fmt" "net/http" "net/http/httptest" "os" @@ -11,6 +14,7 @@ import ( "strings" "testing" + "github.com/Kocoro-lab/ShanClaw/internal/agent" "github.com/Kocoro-lab/ShanClaw/internal/client" ctxwin "github.com/Kocoro-lab/ShanClaw/internal/context" "github.com/Kocoro-lab/ShanClaw/internal/session" @@ -174,13 +178,28 @@ func TestConversationContextCapabilityAdvertised(t *testing.T) { t.Fatalf("Capabilities missing %q", CapConversationContextActionsV1) } -func TestSideChatEndpointUsesReadOnlyEphemeralContext(t *testing.T) { +// sideChatProbeTool is a minimal approval-requiring tool for pinning that +// side chats run with the normal registry + approval flow. +type sideChatProbeTool struct{ calls int } + +func (t *sideChatProbeTool) Info() agent.ToolInfo { + return agent.ToolInfo{Name: "sidechat_probe", Description: "side chat probe tool"} +} +func (t *sideChatProbeTool) RequiresApproval() bool { return true } +func (t *sideChatProbeTool) Run(context.Context, string) (agent.ToolResult, error) { + t.calls++ + return agent.ToolResult{Content: "probe ok"}, nil +} + +func TestSideChatEndpointRunsToolEnabledEphemeralContext(t *testing.T) { gateway := &fakeGatewayBackend{reply: "focused answer"} gatewayServer := httptest.NewServer(gateway.handler()) defer gatewayServer.Close() deps := runAgentContractTestDeps(t, gatewayServer.URL) defer deps.SessionCache.CloseAll() + deps.Registry.Register(&sideChatProbeTool{}) + deps.BaselineReg.Register(&sideChatProbeTool{}) mgr := deps.SessionCache.GetOrCreateManager(deps.SessionCache.SessionsDir("")) source := mgr.NewSessionWithID("side-chat-source") source.CWD = t.TempDir() @@ -214,8 +233,19 @@ func TestSideChatEndpointUsesReadOnlyEphemeralContext(t *testing.T) { } var transcript strings.Builder for _, request := range requests { - if len(request.Tools) != 0 { - t.Fatalf("side chat exposed %d tools", len(request.Tools)) + // Side chats expose the normal registry — same capability as the + // primary conversation, only the lifecycle is ephemeral. + if len(request.Messages) == 0 { + continue // tool-less internal probe call, not a model turn + } + found := false + for _, tool := range request.Tools { + if tool.Function.Name == "sidechat_probe" { + found = true + } + } + if !found { + t.Fatalf("side chat request lost the registered tool: %d tools", len(request.Tools)) } for _, message := range request.Messages { transcript.WriteString(message.Content.Text()) @@ -343,3 +373,144 @@ func TestSideChatUsesCompactionCheckpointHistory(t *testing.T) { t.Fatalf("side chat re-fed the raw archive past the checkpoint: %s", transcript.String()) } } + +func TestShouldInjectQuestionAskerSkipsEphemeralRuns(t *testing.T) { + if !shouldInjectQuestionAsker(RunAgentRequest{Source: "desktop"}) { + t.Fatal("attended desktop run must get an asker") + } + // Side chats are ephemeral panels with no question UI: an asker would + // block for the auto-resolution window then report a phantom decline. + if shouldInjectQuestionAsker(RunAgentRequest{Source: "desktop", Ephemeral: true}) { + t.Fatal("ephemeral run must not get an asker") + } + if shouldInjectQuestionAsker(RunAgentRequest{Source: "slack"}) { + t.Fatal("messaging source must not get an asker") + } +} + +// TestSideChatSSEEmitsApprovalRequestAndRunsTool drives the full loop over a +// live HTTP server: the model calls an approval-requiring tool in a side chat, +// the per-request SSE stream must carry the approval frame, POST /approval +// resolves it, and the tool actually runs. +func TestSideChatSSEEmitsApprovalRequestAndRunsTool(t *testing.T) { + // Streamed SSE done-frames so the loop's streaming path succeeds; the + // response is chosen by transcript state (tool_result present or not) so + // retries and probe calls cannot desynchronize a call counter. + gatewayServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req client.CompletionRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + resp := client.CompletionResponse{Provider: "anthropic", Model: "test-model"} + hasToolResult := false + for _, message := range req.Messages { + for _, block := range message.Content.Blocks() { + if block.Type == "tool_result" { + hasToolResult = true + } + } + } + if hasToolResult { + resp.FinishReason = "end_turn" + resp.OutputText = "probe finished" + } else { + resp.FinishReason = "tool_use" + resp.ToolCalls = []client.FunctionCall{{ + ID: "call-probe-1", Name: "sidechat_probe", + Arguments: json.RawMessage(`{"description":"run the probe"}`), + }} + } + payload, _ := json.Marshal(struct { + Type string `json:"type"` + client.CompletionResponse + }{Type: "done", CompletionResponse: resp}) + w.Header().Set("Content-Type", "text/event-stream") + _, _ = fmt.Fprintf(w, "data: %s\n\ndata: [DONE]\n\n", payload) + if flusher, ok := w.(http.Flusher); ok { + flusher.Flush() + } + })) + defer gatewayServer.Close() + + deps := runAgentContractTestDeps(t, gatewayServer.URL) + defer deps.SessionCache.CloseAll() + probe := &sideChatProbeTool{} + deps.Registry.Register(probe) + deps.BaselineReg.Register(probe) + mgr := deps.SessionCache.GetOrCreateManager(deps.SessionCache.SessionsDir("")) + source := mgr.NewSessionWithID("side-chat-approval-source") + source.Messages = []client.Message{ + {Role: "user", Content: client.NewTextContent("original question")}, + {Role: "assistant", Content: client.NewTextContent("original answer")}, + } + if err := mgr.Save(); err != nil { + t.Fatal(err) + } + + server := NewServer(0, nil, deps, "test") + httpServer := httptest.NewServer(server.Handler()) + defer httpServer.Close() + + body := `{"message_index":2,"text":"run the probe"}` + req, err := http.NewRequest(http.MethodPost, httpServer.URL+"/sessions/side-chat-approval-source/side-chat", strings.NewReader(body)) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "text/event-stream") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d", resp.StatusCode) + } + + sawApproval := false + sawDone := false + scanner := bufio.NewScanner(resp.Body) + scanner.Buffer(make([]byte, 1<<20), 1<<20) + event := "" + for scanner.Scan() { + line := scanner.Text() + switch { + case strings.HasPrefix(line, "event: "): + event = strings.TrimPrefix(line, "event: ") + case strings.HasPrefix(line, "data: "): + data := strings.TrimPrefix(line, "data: ") + switch event { + case "approval": + sawApproval = true + var frame struct { + RequestID string `json:"request_id"` + Tool string `json:"tool"` + } + if err := json.Unmarshal([]byte(data), &frame); err != nil || frame.Tool != "sidechat_probe" { + t.Fatalf("approval frame = %s err=%v", data, err) + } + resolve := fmt.Sprintf(`{"request_id":%q,"decision":"allow"}`, frame.RequestID) + resolveResp, err := http.Post(httpServer.URL+"/approval", "application/json", strings.NewReader(resolve)) + if err != nil { + t.Fatal(err) + } + resolveResp.Body.Close() + if resolveResp.StatusCode != http.StatusOK { + t.Fatalf("resolve status = %d", resolveResp.StatusCode) + } + case "done": + sawDone = true + if !strings.Contains(data, "probe finished") { + t.Fatalf("done frame = %s", data) + } + } + } + } + if !sawApproval || !sawDone { + t.Fatalf("stream missing frames: approval=%t done=%t", sawApproval, sawDone) + } + if probe.calls == 0 { + t.Fatal("approved probe tool never ran") + } +} diff --git a/internal/daemon/runner.go b/internal/daemon/runner.go index 0214f951..4b5ff025 100644 --- a/internal/daemon/runner.go +++ b/internal/daemon/runner.go @@ -1425,6 +1425,16 @@ func CanPresentQuestionUI(source string) bool { return ok } +// shouldInjectQuestionAsker gates the SSE handler's QuestionAsker injection. +// Ephemeral runs (side chat) come from panels with no question UI: an asker +// there blocks the run for the whole auto-resolution window and then reports +// a decline the user never made — the same failure mode that once hit the +// IM channels. Without an asker, ask_user_question degrades to its clean +// "can't ask here" result. +func shouldInjectQuestionAsker(req RunAgentRequest) bool { + return !req.Ephemeral && CanPresentQuestionUI(req.Source) +} + // cacheSourceFromDaemonSource normalizes daemon-level origins for Cloud-side // attribution. It does not select a TTL: Cloud currently applies the short // prompt-cache TTL to every source. See docs/cache-strategy.md. diff --git a/internal/daemon/server.go b/internal/daemon/server.go index 79ee530e..191dcbe5 100644 --- a/internal/daemon/server.go +++ b/internal/daemon/server.go @@ -3517,7 +3517,7 @@ func (s *Server) handleMessageSSE(w http.ResponseWriter, r *http.Request, req Ru // through Cloud, questions do not, so an asker there blocks the run for the // whole auto-resolution window and then reports a decline the user never made. askCtx := r.Context() - if CanPresentQuestionUI(req.Source) { + if shouldInjectQuestionAsker(req) { askCtx = agent.WithQuestionAsker(askCtx, &brokerQuestionAsker{ broker: qBroker, metaFn: func() ApprovalRequestMeta { From 9d0725baa406da4cda14b8f25b592029a9fa72f1 Mon Sep 17 00:00:00 2001 From: Shenghao Hu Date: Tue, 18 Aug 2026 09:32:01 +0900 Subject: [PATCH 4/4] =?UTF-8?q?fix:=20address=20Codex=20review=20=E2=80=94?= =?UTF-8?q?=20deterministic=20envelope=20boundaries,=20fixture=20and=20doc?= =?UTF-8?q?=20debt?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round-2 review fixes for the conversation context actions PR. - Replace stripJoinedConversationReplies' blank-line segment heuristic with the only boundary that is provable by construction: the envelope run immediately after the "[New message from user]" inject prefix (or the head of an injected block batch's leading text block). A well-formed pair at a "\n\n" boundary deeper in the joined text is indistinguishable from the user's own prose paragraph and previously got eaten; it now stays verbatim. Text and block paths share the identical rule; a later follow-up's envelope in a merged batch remains visible (lossless) and its annotations were still delivered to the model. - Pin message_meta[].conversation_annotations through the real GET /sessions/{id} producer: annotated tail message in http_get.session.response.json, strict consumer decode assertion in TestWireFixture_HTTPDefaultSessionDetail. Desktop's vendored fixture copy needs the usual one-way re-sync when it targets this daemon version. - Add Hardcoded Limit Policy declarations (workload / binding symptom / override path) to the three reply-envelope caps and maxSideChatHistoryMessages. All four stay consts, not viper: they are two-sided wire-contract values Desktop hardcodes too, so lifting one is a coordinated release, never a per-machine config edit. - Doc co-maintenance: README.md gains the user-facing annotate / side chat / branch paragraph; CLAUDE.md gains the conversation-context-actions subsystem block (Desktop-only, out of kocoro skill references); AGENTS.md gains the compact wire-contract entry with equal-byte prose trims to stay under the CI-asserted 24576B ceiling (24572B). - Correct the stale "tool-free" claim in the desktop-wire-fixtures README: side chats run the normal tool registry and approval flow; only the lifecycle is ephemeral. Verification: go test ./... passes (full suite, exit 0); go test -race passes on the affected internal/daemon conversation/fixture tests and internal/agent. New regression: prefix-adjacent envelope stripped, mid-join envelope kept verbatim, block-path head envelope handled identically (TestInjectedTurnStripsOnlyThePrefixAdjacentEnvelope). --- AGENTS.md | 38 ++++----- CLAUDE.md | 4 + README.md | 2 + docs/desktop-wire-fixtures/README.md | 2 +- .../http_get.session.response.json | 8 +- internal/daemon/conversation_context.go | 7 ++ internal/daemon/conversation_reply.go | 78 +++++++------------ internal/daemon/conversation_reply_test.go | 37 +++++++++ internal/daemon/wire_fixtures_test.go | 18 ++++- 9 files changed, 119 insertions(+), 75 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 6d9e4d3f..071bce5f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,21 +1,18 @@ # Kocoro Project Guide (AGENTS.md) -**Condensed mirror of `CLAUDE.md` for external coding agents. `CLAUDE.md` in this -directory is the full guide** — open it for reasoning, wire details, incident -history, or any subsystem not listed here. This file carries only rules you can -act on, plus the symbols and constants to grep. If the two disagree, `CLAUDE.md` -and the code win. - -**Keep this file under 24 KB.** Harnesses that read `AGENTS.md` do so under a byte -budget shared across every such file from the repo root down (32 KiB by default), -and an over-budget file is truncated **from the tail, with no marker in the injected -text** — the reader cannot tell it got a partial file, and the sections at the bottom -are simply gone. CI asserts the ceiling; if you need more room, cut prose, not rules. +**Condensed mirror of `CLAUDE.md` for external coding agents; `CLAUDE.md` is the +full guide** for reasoning, wire details, and unlisted subsystems. This file is +actionable rules plus the symbols to grep. If the two disagree, `CLAUDE.md` and +the code win. + +**Keep this file under 24 KB (CI asserts).** Harnesses inject `AGENTS.md` under a +shared 32 KiB budget and truncate over-budget files **silently from the tail** — +bottom sections vanish. Need room? Cut prose, not rules. Kocoro is the Go CLI/runtime (`shan`) for Shannon AI agents. Production path: -daemon + Kocoro Desktop + Shannon Cloud — the daemon holds a Cloud WebSocket, -receives channel messages, runs the agent loop locally with full tool access, and -streams back. Also TUI, one-shot CLI, MCP server, local scheduled tasks. +daemon + Kocoro Desktop + Shannon Cloud — the daemon holds a Cloud WS, receives +channel messages, runs the agent loop locally, streams back. Also TUI, one-shot +CLI, MCP server, schedules. Layout: `cmd/` (Cobra) + `internal//`; use Glob/Grep. Production path is `internal/daemon/` driving `internal/agent/`. @@ -180,10 +177,15 @@ hard-block -> denied commands -> compound splitting -> always-ask gates - `ApprovalBroker` and `QuestionBroker` are thin faces over ONE shared `pendingCore[D]` (`pending.go`). A third interaction kind MUST build on it — do NOT copy a broker. -- `ask_user_question` gates on `CanPresentQuestionUI` / `questionUISources`, an - ALLOW-list, NOT the approval predicate: every source without a question UI - DECLINES, because a question has no safe auto-answer. Slack/Feishu/Lark/Teams/ - LINE render approval cards but have NO question channel. Capability `question_v1`. +- `ask_user_question` gates on `CanPresentQuestionUI` / `questionUISources` — an + ALLOW-list, NOT the approval predicate: sources without a question UI DECLINE + (no safe auto-answer; IM channels have approval cards but no question channel). + Capability `question_v1`. Ephemeral runs get no asker (`shouldInjectQuestionAsker`). +- Conversation context actions (`conversation_context_actions_v1`, Desktop-only): + `POST /sessions/{id}/{fork,side-chat}`; `message_index` = RAW-archive turn + boundary (injected entries counted). Side chats run the NORMAL tool registry + + SSE approvals but stay ephemeral (no session, no bus events). Reply envelopes + strip head-only, only when they parse; limits 400 at `/message` + `/queue`. - `delivery_ack`: ack an inbound message only AFTER reply delivery succeeds. Reply-failure paths skip the ack so replay stays correct. diff --git a/CLAUDE.md b/CLAUDE.md index f3ca8be1..c07674ff 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -159,6 +159,10 @@ One `####` per subsystem. Each names its code home first, then the invariant. `daemon/runner.go` + `daemon/skill_recommendation.go`. The two tools above are Direct only for capable signed-in Desktop/Kocoro requests with the feature enabled; absent from TUI, one-shot, MCP, schedule, IM, heartbeat, and watcher runs. A disconnected card-event sink does NOT change schemas — `offer_skill_installation` fails closed when invoked. +#### Conversation context actions (Branch / Side Chat / reply annotations) + +`internal/daemon/conversation_context.go` + `conversation_reply.go`, capability `conversation_context_actions_v1`, Desktop-only transport (the agent never calls these → NOT in kocoro skill references; wire contract in `docs/desktop-wire-fixtures/` + `docs/cloud-contract-surface.md`). `POST /sessions/{id}/fork` copies history through a complete assistant turn into a normal persisted session (optional `target_agent`, validated to exist); `POST /sessions/{id}/side-chat` runs an ephemeral turn over that bounded history. `message_index` on both is a RAW-archive boundary (system-injected entries counted) — the same index space Desktop's `rawIndex` uses. Both honor a covering `CompactionCheckpoint` (fork deep-copies it; side-chat feeds checkpoint+tail via `Session.HistoryThrough`). Side chats run the NORMAL tool registry + permission engine with SSE approvals, but stay ephemeral: no session persisted, no bus events, and no question asker (`shouldInjectQuestionAsker` — the panel has no question UI, so an asker would block then report a phantom decline). Desktop text replies ride a transient head-only `` envelope: stripped from the archive ONLY when it parses (malformed stays verbatim — lossless), decoded into `message_meta[].conversation_annotations`, limits (100 replies / 8K quote / 2K comment runes) enforced with stable 400 codes at `/message` + `/queue`. + #### WS handshake `client.go`. Sends `User-Agent: kocoro/` + `X-Kocoro-Daemon-Version` + `X-Kocoro-Capabilities`. diff --git a/README.md b/README.md index 76a86e98..e247128c 100644 --- a/README.md +++ b/README.md @@ -15,6 +15,8 @@ Kocoro runs AI agents locally with full computer access — files, apps, browser > **Coming from Claude Code?** Kocoro Desktop can import your existing agents, skills, and instructions from `~/.claude/` in one click — preview-then-apply via the daemon's `/migrate/claude-code/*` endpoints. +In Kocoro Desktop you can also work *inside* a conversation: select any passage of a reply to annotate it with inline comments (they reach the agent alongside your next message and survive reload), open a **Side Chat** — a temporary aside with the same tools and approvals as the main conversation that vanishes when the app closes — or **branch** a finished turn into a new session, optionally under a different agent. All three are served by this daemon (`conversation_context_actions_v1`). + Built on **[Shannon](https://github.com/Kocoro-lab/Shannon)** — the open-source multi-agent framework that powers both the Shannon Cloud SaaS and the self-hosted Shannon Gateway. ## Contents diff --git a/docs/desktop-wire-fixtures/README.md b/docs/desktop-wire-fixtures/README.md index d17746c5..1f598c68 100644 --- a/docs/desktop-wire-fixtures/README.md +++ b/docs/desktop-wire-fixtures/README.md @@ -149,7 +149,7 @@ codecs. | `http_get.sessions.schedule.response.json` | `server.go handleSessions` (`GET /sessions?schedule_id=`) | exact scheduled-task session filter. Each matching row carries persistent `schedule_id`; deleting the schedule configuration does not delete or rewrite the session. | | `http_get.session.response.json` | `server.go handleGetSession` (`GET /sessions/{id}`) | lossless default session detail. Top-level `messages` remain the archive; `compaction_checkpoint` additively exposes the durable model-live state and its exclusive raw archive index. No new capability token is required: clients that ignore the additive field retain the existing archive semantics; consumers that opt into live-context metrics read the checkpoint explicitly. | | `http_get.session.remote_timeline.response.json` | `server.go handleGetSession` (`GET /sessions/{id}?view=remote_timeline`) | capability-gated mobile projection of the archive. Returns a byte-bounded newest page with aligned `messages` / `message_meta`, absolute `start_index`, opaque `next_cursor`, `has_more`, and explicit `omitted_content_count`; it projects from top-level archived `messages`, not the model-live checkpoint. | -| `http_post.session_fork.response.json` | `conversation_context.go handleForkSession` (`POST /sessions/{id}/fork`) | fork response for `conversation_context_actions_v1`. `session_id` is the freshly generated fork id (shape-asserted, normalized in the test); `title` snapshots the source session's title. The side-chat route has no fixture of its own: its non-SSE response is a `RunAgentResult` and its SSE stream reuses the already-pinned per-request frames; the boundary, tool-free, and non-persistence contracts are exercised in `internal/daemon/conversation_context_test.go`. | +| `http_post.session_fork.response.json` | `conversation_context.go handleForkSession` (`POST /sessions/{id}/fork`) | fork response for `conversation_context_actions_v1`. `session_id` is the freshly generated fork id (shape-asserted, normalized in the test); `title` snapshots the source session's title. The side-chat route has no fixture of its own: its non-SSE response is a `RunAgentResult` and its SSE stream reuses the already-pinned per-request frames (including `approval` — side chats run the NORMAL tool registry and approval flow; only the lifecycle is ephemeral). The boundary, full-tools + approval, and non-persistence contracts are exercised in `internal/daemon/conversation_context_test.go`. | ### Quick-panel surfaces (POST request bodies + error responses) diff --git a/docs/desktop-wire-fixtures/http_get.session.response.json b/docs/desktop-wire-fixtures/http_get.session.response.json index f09a8a48..8962f58d 100644 --- a/docs/desktop-wire-fixtures/http_get.session.response.json +++ b/docs/desktop-wire-fixtures/http_get.session.response.json @@ -48,7 +48,13 @@ }, { "source": "desktop", - "timestamp": "2026-08-05T00:00:03Z" + "timestamp": "2026-08-05T00:00:03Z", + "conversation_annotations": [ + { + "selected_text": "ARCHIVE_ONLY_OLD_REPLY", + "comment": "explain this line" + } + ] } ], "source": "desktop", diff --git a/internal/daemon/conversation_context.go b/internal/daemon/conversation_context.go index 64a2d62d..9440fe06 100644 --- a/internal/daemon/conversation_context.go +++ b/internal/daemon/conversation_context.go @@ -13,6 +13,13 @@ import ( "github.com/Kocoro-lab/ShanClaw/internal/session" ) +// maxSideChatHistoryMessages bounds the panel-supplied temporary transcript on +// POST /sessions/{id}/side-chat. Workload: a side chat is a short focused +// aside — tens of turns, not hundreds; 100 is several times the longest panel +// session observed in testing. When it binds the route 400s ("side chat +// history is too long") and Desktop trims its oldest panel turns client-side. +// Const, not viper: the bound is a request-size guard on a Desktop-only wire +// surface, so lifting it is a coordinated Desktop + daemon change. const maxSideChatHistoryMessages = 100 type forkSessionRequest struct { diff --git a/internal/daemon/conversation_reply.go b/internal/daemon/conversation_reply.go index d89b98b0..7e36c59f 100644 --- a/internal/daemon/conversation_reply.go +++ b/internal/daemon/conversation_reply.go @@ -13,8 +13,20 @@ import ( ) const ( - conversationRepliesOpen = "" - conversationRepliesClose = "" + conversationRepliesOpen = "" + conversationRepliesClose = "" + + // Wire-contract caps for one message's reply envelope, mirrored by + // Desktop's composer (its textarea maxlength and annotation editor use the + // same numbers). Workload: one reply per selected passage — 100 covers a + // user annotating every paragraph of a very long assistant turn, 8K runes + // a full-screen text selection, 2K runes a paragraph-length comment. When + // they bind, the Desktop ingresses 400 with a stable + // conversation_replies_* / conversation_reply_* code; paths that cannot + // reject truncate only the persisted display metadata (the model still + // sees the full envelope). Deliberately consts, not viper: both ends of + // the contract hardcode the values, so lifting them is a coordinated + // Desktop + daemon release, never a per-machine config edit. maxConversationAnnotations = 100 maxConversationAnnotationQuoteRunes = 8_000 maxConversationAnnotationCommentRunes = 2_000 @@ -206,15 +218,16 @@ func conversationReplyPersistedMessage(message client.Message) (client.Message, // conversationReplyVisibleText splits + decodes a head envelope. stripped is // false (and visible == text) when there is no envelope or it fails to parse. // An injected-turn message (the agent loop's "[New message from user]" join of -// drained follow-ups) is handled segment-aware, because each follow-up's -// envelope sits at that follow-up's head inside the joined text, not at the -// message head. +// drained follow-ups) is handled by stripping the envelope run immediately +// after that prefix — the only position that is a follow-up head BY +// CONSTRUCTION. Positions further into the joined text are ambiguous (a +// "\n\n" boundary can equally be a paragraph of the user's own prose), so a +// later follow-up's envelope stays verbatim rather than risk eating user +// bytes; its annotations were still delivered to the model, only the reload +// chip for that rare merged batch is missing. func conversationReplyVisibleText(text string) (visible string, annotations []session.ConversationAnnotation, stripped bool) { - if rest, ok := strings.CutPrefix(text, agent.InjectedUserMessagePrefix); ok { - visible, annotations, stripped = stripJoinedConversationReplies(rest) - return agent.InjectedUserMessagePrefix + visible, annotations, stripped - } - split, envelope := splitConversationReplyPrompt(text) + rest, injected := strings.CutPrefix(text, agent.InjectedUserMessagePrefix) + split, envelope := splitConversationReplyPrompt(rest) if envelope == "" { return text, nil, false } @@ -222,47 +235,8 @@ func conversationReplyVisibleText(text string) (visible string, annotations []se if annotations == nil { return text, nil, false } - return split, annotations, true -} - -// stripJoinedConversationReplies removes every well-formed envelope run that -// starts at the head of a "\n\n"-joined segment — the positions where a -// drained follow-up's own head can land inside an injected-turn message. -// Anything that fails to parse, or sits anywhere else, stays verbatim. -func stripJoinedConversationReplies(text string) (string, []session.ConversationAnnotation, bool) { - var out strings.Builder - var annotations []session.ConversationAnnotation - stripped := false - i := 0 - for { - idx := strings.Index(text[i:], conversationRepliesOpen) - if idx < 0 { - out.WriteString(text[i:]) - break - } - abs := i + idx - atSegmentHead := abs == 0 || strings.HasSuffix(text[:abs], "\n\n") - if atSegmentHead { - segVisible, envelope := splitConversationReplyPrompt(text[abs:]) - if decoded := conversationReplyAnnotations(envelope); envelope != "" && decoded != nil { - out.WriteString(text[:abs][i:]) - annotations = append(annotations, decoded...) - if len(annotations) > maxConversationAnnotations { - annotations = annotations[:maxConversationAnnotations] - } - stripped = true - // splitConversationReplyPrompt already trimmed the separator - // between the envelope run and the segment's visible text. - text = segVisible - i = 0 - continue - } - } - out.WriteString(text[i : abs+len(conversationRepliesOpen)]) - i = abs + len(conversationRepliesOpen) - } - if !stripped { - return text, nil, false + if injected { + split = agent.InjectedUserMessagePrefix + split } - return out.String(), annotations, true + return split, annotations, true } diff --git a/internal/daemon/conversation_reply_test.go b/internal/daemon/conversation_reply_test.go index c16aaf19..e13cecab 100644 --- a/internal/daemon/conversation_reply_test.go +++ b/internal/daemon/conversation_reply_test.go @@ -64,6 +64,43 @@ func TestConversationReplyPersistedMessageStripsHeadEnvelopeOnly(t *testing.T) { } } +func TestInjectedTurnStripsOnlyThePrefixAdjacentEnvelope(t *testing.T) { + prefix := agent.InjectedUserMessagePrefix + env := "q1c1" + + // The envelope immediately after the inject prefix is a follow-up head by + // construction and is stripped. + head, annotations := conversationReplyPersistedMessage(client.Message{ + Role: "user", Content: client.NewTextContent(prefix + env + "\n\nfollow-up text"), + }) + if head.Content.Text() != prefix+"follow-up text" || len(annotations) != 1 { + t.Fatalf("prefix-adjacent envelope not stripped: %q %#v", head.Content.Text(), annotations) + } + + // A well-formed pair at a "\n\n" boundary deeper into the joined text is + // indistinguishable from the user's own prose paragraph — it must stay + // verbatim (the old blank-line heuristic ate it). + prose := prefix + "please explain this snippet:\n\n" + env + "\n\nthanks" + kept, keptAnnotations := conversationReplyPersistedMessage(client.Message{ + Role: "user", Content: client.NewTextContent(prose), + }) + if kept.Content.Text() != prose || keptAnnotations != nil { + t.Fatalf("mid-join envelope was eaten: %q %#v", kept.Content.Text(), keptAnnotations) + } + + // Block-path injected batches (files present) carry no prefix; the head + // envelope of the first follow-up gets the identical treatment. + blockMsg, blockAnnotations := conversationReplyPersistedMessage(client.Message{ + Role: "user", Content: client.NewBlockContent([]client.ContentBlock{ + {Type: "text", Text: env + "\n\nsee attachment"}, + {Type: "image", Source: &client.ImageSource{Type: "base64", MediaType: "image/png", Data: "abc"}}, + }), + }) + if blockMsg.Content.Blocks()[0].Text != "see attachment" || len(blockAnnotations) != 1 { + t.Fatalf("block-path head envelope not stripped: %#v", blockMsg.Content.Blocks()) + } +} + func TestConversationReplyPersistedMessageKeepsMalformedEnvelopeVerbatim(t *testing.T) { malformed := "broken\n\nquery" kept, annotations := conversationReplyPersistedMessage(client.Message{ diff --git a/internal/daemon/wire_fixtures_test.go b/internal/daemon/wire_fixtures_test.go index 96f8b6d5..c824fed1 100644 --- a/internal/daemon/wire_fixtures_test.go +++ b/internal/daemon/wire_fixtures_test.go @@ -1592,7 +1592,9 @@ func TestWireFixture_HTTPDefaultSessionDetail(t *testing.T) { sess.MessageMeta = []session.MessageMeta{ {Source: "desktop", Timestamp: &firstTime}, {Source: "desktop", Timestamp: &secondTime}, - {Source: "desktop", Timestamp: &tailTime}, + {Source: "desktop", Timestamp: &tailTime, ConversationAnnotations: []session.ConversationAnnotation{ + {SelectedText: "ARCHIVE_ONLY_OLD_REPLY", Comment: "explain this line"}, + }}, } sess.CompactionCheckpoint = &session.CompactionCheckpoint{ SchemaVersion: session.CompactionCheckpointSchemaVersion, @@ -1643,8 +1645,12 @@ func TestWireFixture_HTTPDefaultSessionDetail(t *testing.T) { Content json.RawMessage `json:"content"` } `json:"messages"` MessageMeta []struct { - Source string `json:"source"` - Timestamp string `json:"timestamp"` + Source string `json:"source"` + Timestamp string `json:"timestamp"` + ConversationAnnotations []struct { + SelectedText string `json:"selected_text"` + Comment string `json:"comment"` + } `json:"conversation_annotations"` } `json:"message_meta"` CompactionCheckpoint *struct { SchemaVersion int `json:"schema_version"` @@ -1684,6 +1690,12 @@ func TestWireFixture_HTTPDefaultSessionDetail(t *testing.T) { t.Fatalf("archive/live checkpoint semantics drifted: archive=%s checkpoint=%s", detail.Messages[0].Content, cp.Messages[1].Content) } + tailAnnotations := detail.MessageMeta[2].ConversationAnnotations + if len(tailAnnotations) != 1 || tailAnnotations[0].SelectedText != "ARCHIVE_ONLY_OLD_REPLY" || + tailAnnotations[0].Comment != "explain this line" || + len(detail.MessageMeta[0].ConversationAnnotations) != 0 { + t.Fatalf("consumer decode lost conversation annotations: %+v", detail.MessageMeta) + } } func TestWireFixture_HTTPRemoteSessionTimeline(t *testing.T) {