diff --git a/apps/backend/AGENTS.md b/apps/backend/AGENTS.md index 9d11def8e3..77a34f56c5 100644 --- a/apps/backend/AGENTS.md +++ b/apps/backend/AGENTS.md @@ -204,6 +204,14 @@ Every long-running goroutine must have a single owner with explicit start and st - After all repos complete `initSchema`, `cmd/kandev/storage.go:recordSchemaVersion` writes the current binary version into `kandev_meta` (non-fatal; a failure just means the next boot will take a fresh snapshot). - Migration logging: `db.MigrateLogger.Apply(name, stmt)` — success logs Info, "already exists" / "duplicate column name" is silently swallowed, anything else logs Warn but never returns an error (preserving the existing swallow-error contract). +## Schema & migrations (SQLite repository) + +`initSchema()` in `internal/task/repository/sqlite/base_schema.go` runs the `init*Schema` (CREATE TABLE) steps **before** `runMigrations()`. The table-creation DDL uses `CREATE TABLE IF NOT EXISTS`, so on an **existing** database it is a no-op and never adds columns to a table that is already present. + +**Rule:** when you add a column to an existing table, add it **only** via an idempotent `ADD COLUMN` migration in `runMigrations()` (`base_migrations.go`), never by editing the table's `CREATE TABLE` alone. Anything that *references* that new column — an index, a backfill `UPDATE`, a partial-index predicate — must live in `runMigrations()` **after** the `ADD COLUMN`, not in the `init*Schema` DDL. Putting a `CREATE INDEX ... (new_col)` in the schema-init block crashes existing DBs with `no such column: new_col`, because schema init runs before the migration that adds the column. + +You may still list the column in the `CREATE TABLE` so fresh DBs get it inline, but the migration is the source of truth for evolution and must stand alone. New columns also need: the struct field in `models/`, the DTO field + `ToAPI` in `pkg/api/v1/`, and every `CreateX`/`UpdateX`/bulk write in the repo that should set it. + ## Code-quality limits Enforced by `apps/backend/.golangci.yml` (errors on new code only): diff --git a/apps/backend/internal/mcp/server/ask_user_question_test.go b/apps/backend/internal/mcp/server/ask_user_question_test.go index b3deddbfd7..201b85da72 100644 --- a/apps/backend/internal/mcp/server/ask_user_question_test.go +++ b/apps/backend/internal/mcp/server/ask_user_question_test.go @@ -1,9 +1,17 @@ package mcp import ( + "bufio" + "context" "encoding/json" + "net/http" + "net/http/httptest" + "strings" "testing" + "testing/synctest" + "time" + "github.com/gin-gonic/gin" ws "github.com/kandev/kandev/pkg/websocket" mcplib "github.com/mark3labs/mcp-go/mcp" "github.com/stretchr/testify/assert" @@ -244,3 +252,149 @@ func TestAskUserQuestion_RejectionPath(t *testing.T) { assert.Contains(t, textBlock.Text, "rejected") assert.Contains(t, textBlock.Text, "User skipped") } + +// TestEmitKeepAlivePings_TicksUntilStop verifies the keepalive loop fires send on +// each interval and exits once the stop channel closes. +func TestEmitKeepAlivePings_TicksUntilStop(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + stop := make(chan struct{}) + calls := 0 + send := func() { + calls++ + if calls == 3 { + close(stop) + } + } + emitKeepAlivePings(context.Background(), stop, 20*time.Second, send) + assert.Equal(t, 3, calls) + }) +} + +// TestEmitKeepAlivePings_StopsOnContextCancel verifies a cancelled context tears +// the loop down even if stop never closes. +func TestEmitKeepAlivePings_StopsOnContextCancel(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + stop := make(chan struct{}) + defer close(stop) + calls := 0 + send := func() { + calls++ + if calls == 2 { + cancel() + } + } + emitKeepAlivePings(ctx, stop, 20*time.Second, send) + assert.Equal(t, 2, calls) + }) +} + +// TestEmitKeepAlivePings_Guards ensures the loop returns immediately for a +// non-positive interval or a nil send without spinning or panicking. +func TestEmitKeepAlivePings_Guards(t *testing.T) { + emitKeepAlivePings(context.Background(), make(chan struct{}), 0, func() {}) + emitKeepAlivePings(context.Background(), make(chan struct{}), time.Second, nil) +} + +// blockingAskBackend blocks the ask_user_question round-trip until release is +// closed, simulating a user who takes a long time to answer. +type blockingAskBackend struct { + release <-chan struct{} + response map[string]interface{} +} + +func (b *blockingAskBackend) RequestPayload(ctx context.Context, action string, _, result interface{}) error { + if action == ws.ActionMCPAskUserQuestion { + select { + case <-b.release: + case <-ctx.Done(): + return ctx.Err() + } + } + if b.response != nil && result != nil { + data, _ := json.Marshal(b.response) + return json.Unmarshal(data, result) + } + return nil +} + +// TestAskUserQuestion_StreamsKeepAliveDuringWait drives a real Streamable HTTP +// tool call whose backend blocks until the test releases it, and asserts that +// progress notifications stream on the in-flight POST response before the final +// result. This is the regression guard for the agent's MCP client aborting the +// call with "fetch failed" after its idle timeout. +func TestAskUserQuestion_StreamsKeepAliveDuringWait(t *testing.T) { + prev := askQuestionKeepAliveInterval + askQuestionKeepAliveInterval = 5 * time.Millisecond + t.Cleanup(func() { askQuestionKeepAliveInterval = prev }) + + release := make(chan struct{}) + backend := &blockingAskBackend{ + release: release, + response: map[string]interface{}{ + "answers": []interface{}{ + map[string]interface{}{ + "question_id": "q1", + "selected_options": []interface{}{"q1_opt1"}, + }, + }, + }, + } + + log := newTestLogger(t) + s := New(backend, "sess-keepalive", "task-keepalive", 0, log, "", false, ModeTask) + + gin.SetMode(gin.TestMode) + router := gin.New() + s.RegisterRoutes(router) + ts := httptest.NewServer(router) + defer ts.Close() + + initReq := `{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"test","version":"1.0.0"}}}` + initResp := postJSONRPC(t, ts.URL+"/mcp", initReq, "") + require.Equal(t, http.StatusOK, initResp.statusCode, "init: %s", initResp.body) + + callBody := `{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"ask_user_question_kandev","arguments":{"questions":[{"id":"q1","prompt":"Which?","options":[{"label":"A","description":"a"},{"label":"B","description":"b"}]}]}}}` + req, err := http.NewRequest(http.MethodPost, ts.URL+"/mcp", strings.NewReader(callBody)) + require.NoError(t, err) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json, text/event-stream") + req.Header.Set("Mcp-Session-Id", initResp.sessionID) + + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + defer func() { _ = resp.Body.Close() }() + require.Equal(t, http.StatusOK, resp.StatusCode) + + scanner := bufio.NewScanner(resp.Body) + released := false + progressSeen := 0 + finalSeen := false + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if !strings.HasPrefix(line, "data:") { + continue + } + payload := strings.TrimSpace(strings.TrimPrefix(line, "data:")) + switch { + case strings.Contains(payload, "notifications/progress"): + progressSeen++ + if !released { + close(release) + released = true + } + case strings.Contains(payload, `"id":2`): + finalSeen = true + assert.Contains(t, payload, "q1_opt1", "final result should carry the answer") + } + if finalSeen { + break + } + } + require.NoError(t, scanner.Err()) + if !released { + close(release) + } + assert.GreaterOrEqual(t, progressSeen, 1, "expected at least one keepalive progress notification") + assert.True(t, finalSeen, "expected the final tool result to be delivered") +} diff --git a/apps/backend/internal/mcp/server/handlers.go b/apps/backend/internal/mcp/server/handlers.go index 9228ad3a8f..505c6e2841 100644 --- a/apps/backend/internal/mcp/server/handlers.go +++ b/apps/backend/internal/mcp/server/handlers.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "time" ws "github.com/kandev/kandev/pkg/websocket" "github.com/mark3labs/mcp-go/mcp" @@ -11,6 +12,15 @@ import ( "go.uber.org/zap" ) +// askQuestionKeepAliveInterval is how often ask_user_question streams a progress +// notification to the agent while waiting for the user's answer. The agent's MCP +// client (auggie runs on Node, whose fetch/undici applies a 300s idle timeout to +// the in-flight tool-call request) aborts the call with "fetch failed" if no bytes +// arrive for that long. Emitting a progress notification well inside that window +// keeps the streamed POST/SSE response alive so the call survives until the user +// responds. Declared as a var so tests can shorten it. +var askQuestionKeepAliveInterval = 20 * time.Second + // Argument-name constants used across the ask_user_question_kandev handler. // Pulled out so goconst stays happy and renames stay safe. const ( @@ -26,6 +36,7 @@ const ( answeredFieldKey = "answered" rejectedFieldKey = "rejected" documentArg = "document" + messageArg = "message" ) func (s *Server) listWorkspacesHandler() server.ToolHandlerFunc { @@ -320,6 +331,15 @@ func (s *Server) askUserQuestionHandler() server.ToolHandlerFunc { "context": questionCtx, } + // Waiting on a human answer routinely outlasts the agent MCP client's + // idle timeout on the in-flight tool call. Stream periodic progress + // notifications until the backend responds; mcp-go flushes them onto the + // POST/SSE response, resetting the client's idle timer so the call is not + // aborted mid-question. + stop := make(chan struct{}) + defer close(stop) + go emitKeepAlivePings(ctx, stop, askQuestionKeepAliveInterval, s.clarificationKeepAlive(ctx, req)) + // Use the MCP request context from the agent. This ensures that if the agent's // MCP client times out, we'll detect it and not update the session state. var result map[string]interface{} @@ -337,6 +357,54 @@ func (s *Server) askUserQuestionHandler() server.ToolHandlerFunc { } } +// emitKeepAlivePings invokes send on every interval tick until stop is closed or +// ctx is cancelled. It is the transport-agnostic core of the ask_user_question +// keepalive, split out so the timing loop is unit-testable without a live MCP +// session. +func emitKeepAlivePings(ctx context.Context, stop <-chan struct{}, interval time.Duration, send func()) { + if interval <= 0 || send == nil { + return + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-stop: + return + case <-ctx.Done(): + return + case <-ticker.C: + send() + } + } +} + +// clarificationKeepAlive builds the keepalive callback that streams a +// notifications/progress message to the agent. The progress token mirrors the one +// the client attached to the tool call when present, so spec-compliant clients +// associate the updates with the in-flight request; clients that omitted a token +// ignore the unknown one. Returns a no-op when no MCP server is bound to the +// context (e.g. direct unit-test invocation of the handler). +func (s *Server) clarificationKeepAlive(ctx context.Context, req mcp.CallToolRequest) func() { + srv := server.ServerFromContext(ctx) + if srv == nil { + return func() {} + } + var token mcp.ProgressToken = fmt.Sprintf("ask_user_question:%s", s.sessionID) + if req.Params.Meta != nil && req.Params.Meta.ProgressToken != nil { + token = req.Params.Meta.ProgressToken + } + var progress float64 + return func() { + progress++ + _ = srv.SendNotificationToClient(ctx, "notifications/progress", map[string]any{ + "progressToken": token, + "progress": progress, + messageArg: "Waiting for your response in Kandev", + }) + } +} + // parseQuestions extracts and validates the "questions" array from the request. // Returns the normalized question payloads (with auto-assigned ids) on success // or a tool-result error describing the first validation failure. diff --git a/apps/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go index 7e0abe9807..8cbcb02114 100644 --- a/apps/backend/internal/task/models/models.go +++ b/apps/backend/internal/task/models/models.go @@ -445,6 +445,7 @@ type Message struct { Metadata map[string]interface{} `json:"metadata,omitempty"` RequestsInput bool `json:"requests_input"` // True if agent is requesting user input CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` // Authoritative per-message change signal } // ToAPI converts internal Message to API type. @@ -478,6 +479,7 @@ func (m *Message) ToAPI() *v1.Message { Metadata: meta, RequestsInput: m.RequestsInput, CreatedAt: m.CreatedAt, + UpdatedAt: m.UpdatedAt, } if hasHidden { result.RawContent = m.Content diff --git a/apps/backend/internal/task/repository/sqlite/base_migrations.go b/apps/backend/internal/task/repository/sqlite/base_migrations.go index 6f41683646..ecc1dc4a65 100644 --- a/apps/backend/internal/task/repository/sqlite/base_migrations.go +++ b/apps/backend/internal/task/repository/sqlite/base_migrations.go @@ -86,6 +86,15 @@ func (r *Repository) runMigrations() error { r.migrate.Apply("task_sessions.workspace_path", `ALTER TABLE task_sessions ADD COLUMN workspace_path TEXT DEFAULT ''`) r.migrate.Apply("repositories.copy_files", `ALTER TABLE repositories ADD COLUMN copy_files TEXT DEFAULT ''`) + // Authoritative per-message change signal (chat render-perf). SQLite forbids a + // non-constant default on ADD COLUMN, so the column is added nullable and + // existing rows are backfilled to created_at; new inserts/updates set it + // explicitly in CreateMessage/UpdateMessage. The backfill UPDATE is idempotent + // (WHERE updated_at IS NULL). + r.migrate.Apply("task_session_messages.updated_at", `ALTER TABLE task_session_messages ADD COLUMN updated_at TIMESTAMP`) + r.migrate.Apply("task_session_messages.updated_at.backfill", `UPDATE task_session_messages SET updated_at = created_at WHERE updated_at IS NULL`) + r.migrate.Apply("idx_messages_session_updated", `CREATE INDEX IF NOT EXISTS idx_messages_session_updated ON task_session_messages(task_session_id, updated_at)`) + // Backfill the per-session cost/token columns. Runs after the gated // task_sessions rebuilds above so it repairs legacy DBs whose schema can no // longer trigger a rebuild (see migrateSessionsAddCostColumns). diff --git a/apps/backend/internal/task/repository/sqlite/base_schema.go b/apps/backend/internal/task/repository/sqlite/base_schema.go index 26f4b0e738..a8c3d32f4a 100644 --- a/apps/backend/internal/task/repository/sqlite/base_schema.go +++ b/apps/backend/internal/task/repository/sqlite/base_schema.go @@ -465,6 +465,7 @@ func (r *Repository) initMessageTurnSchema() error { type TEXT NOT NULL DEFAULT 'message', metadata TEXT DEFAULT '{}', created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (task_session_id) REFERENCES task_sessions(id) ON DELETE CASCADE, FOREIGN KEY (turn_id) REFERENCES task_session_turns(id) ON DELETE CASCADE ); @@ -473,6 +474,10 @@ func (r *Repository) initMessageTurnSchema() error { CREATE INDEX IF NOT EXISTS idx_messages_created_at ON task_session_messages(created_at); CREATE INDEX IF NOT EXISTS idx_messages_session_created ON task_session_messages(task_session_id, created_at); CREATE INDEX IF NOT EXISTS idx_messages_turn_id ON task_session_messages(turn_id); + -- idx_messages_session_updated is created in runMigrations() after the + -- updated_at ADD COLUMN + backfill. Creating it here would fail on existing + -- DBs where CREATE TABLE IF NOT EXISTS is a no-op and the column does not + -- yet exist (schema init runs before migrations). CREATE TABLE IF NOT EXISTS task_session_turns ( id TEXT PRIMARY KEY, diff --git a/apps/backend/internal/task/repository/sqlite/message.go b/apps/backend/internal/task/repository/sqlite/message.go index 70fffcfde6..afc04cfa4c 100644 --- a/apps/backend/internal/task/repository/sqlite/message.go +++ b/apps/backend/internal/task/repository/sqlite/message.go @@ -24,6 +24,9 @@ func (r *Repository) CreateMessage(ctx context.Context, message *models.Message) if message.CreatedAt.IsZero() { message.CreatedAt = time.Now().UTC() } + if message.UpdatedAt.IsZero() { + message.UpdatedAt = message.CreatedAt + } if message.AuthorType == "" { message.AuthorType = models.MessageAuthorUser } @@ -48,9 +51,9 @@ func (r *Repository) CreateMessage(ctx context.Context, message *models.Message) } _, err := r.db.ExecContext(ctx, r.db.Rebind(` - INSERT INTO task_session_messages (id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - `), message.ID, message.TaskSessionID, message.TaskID, message.TurnID, message.AuthorType, message.AuthorID, message.Content, requestsInput, messageType, metadataJSON, message.CreatedAt) + INSERT INTO task_session_messages (id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `), message.ID, message.TaskSessionID, message.TaskID, message.TurnID, message.AuthorType, message.AuthorID, message.Content, requestsInput, messageType, metadataJSON, message.CreatedAt, message.UpdatedAt) return err } @@ -62,9 +65,9 @@ func (r *Repository) GetMessage(ctx context.Context, id string) (*models.Message var messageType string var metadataJSON string err := r.ro.QueryRowContext(ctx, r.ro.Rebind(` - SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at + SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at FROM task_session_messages WHERE id = ? - `), id).Scan(&message.ID, &message.TaskSessionID, &message.TaskID, &message.TurnID, &message.AuthorType, &message.AuthorID, &message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt) + `), id).Scan(&message.ID, &message.TaskSessionID, &message.TaskID, &message.TurnID, &message.AuthorType, &message.AuthorID, &message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt, &message.UpdatedAt) if err != nil { return nil, err } @@ -83,7 +86,7 @@ func (r *Repository) GetMessage(ctx context.Context, id string) (*models.Message // ListMessages returns all messages for a session ordered by creation time. func (r *Repository) ListMessages(ctx context.Context, sessionID string) ([]*models.Message, error) { rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(` - SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at + SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at FROM task_session_messages WHERE task_session_id = ? ORDER BY created_at ASC `), sessionID) if err != nil { @@ -99,7 +102,7 @@ func (r *Repository) ListMessages(ctx context.Context, sessionID string) ([]*mod // own rows rather than the whole session's history. func (r *Repository) ListMessagesByTurnID(ctx context.Context, turnID string) ([]*models.Message, error) { rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(` - SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at + SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at FROM task_session_messages WHERE turn_id = ? ORDER BY created_at ASC `), turnID) if err != nil { @@ -162,7 +165,7 @@ func (r *Repository) resolveMessageCursor(ctx context.Context, sessionID string, func buildListMessagesQuery(sessionID string, opts models.ListMessagesOptions, cursor *models.Message, sortDir string, limit int) (string, []interface{}) { query := ` - SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at + SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at FROM task_session_messages WHERE task_session_id = ?` args := []interface{}{sessionID} if cursor != nil { @@ -192,7 +195,7 @@ func scanMessageRows(rows interface { var requestsInput int var messageType string var metadataJSON string - if err := rows.Scan(&message.ID, &message.TaskSessionID, &message.TaskID, &message.TurnID, &message.AuthorType, &message.AuthorID, &message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt); err != nil { + if err := rows.Scan(&message.ID, &message.TaskSessionID, &message.TaskID, &message.TurnID, &message.AuthorType, &message.AuthorID, &message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt, &message.UpdatedAt); err != nil { return nil, false, err } message.RequestsInput = requestsInput == 1 @@ -233,7 +236,7 @@ func (r *Repository) SearchMessages(ctx context.Context, sessionID string, opts escaper := strings.NewReplacer(`\`, `\\`, `%`, `\%`, `_`, `\_`) pattern := "%" + escaper.Replace(query) + "%" sql := fmt.Sprintf(` - SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at + SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at FROM task_session_messages WHERE task_session_id = ? AND content %s ? ESCAPE '\' ORDER BY created_at DESC, id DESC @@ -269,12 +272,12 @@ func (r *Repository) getMessageByMetadataField(ctx context.Context, sessionID, f var metadataJSON string drv := r.ro.DriverName() query := fmt.Sprintf(` - SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at + SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at FROM task_session_messages WHERE task_session_id = ? AND %s = ? %s `, dialect.JSONExtract(drv, "metadata", fieldName), orderSuffix) err := r.ro.QueryRowContext(ctx, r.ro.Rebind(query), sessionID, fieldValue).Scan(&message.ID, &message.TaskSessionID, &message.TaskID, &message.TurnID, &message.AuthorType, &message.AuthorID, - &message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt) + &message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt, &message.UpdatedAt) if err != nil { return nil, err } @@ -315,12 +318,12 @@ func (r *Repository) FindMessageByPendingID(ctx context.Context, pendingID strin var metadataJSON string drv := r.ro.DriverName() query := fmt.Sprintf(` - SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at + SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at FROM task_session_messages WHERE %s = ? ORDER BY created_at DESC LIMIT 1 `, dialect.JSONExtract(drv, "metadata", "pending_id")) err := r.ro.QueryRowContext(ctx, r.ro.Rebind(query), pendingID).Scan(&message.ID, &message.TaskSessionID, &message.TaskID, &message.TurnID, &message.AuthorType, &message.AuthorID, - &message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt) + &message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt, &message.UpdatedAt) if err != nil { return nil, err } @@ -340,7 +343,7 @@ func (r *Repository) FindMessageByPendingID(ctx context.Context, pendingID strin func (r *Repository) FindMessagesByPendingID(ctx context.Context, pendingID string) ([]*models.Message, error) { drv := r.ro.DriverName() query := fmt.Sprintf(` - SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at + SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at FROM task_session_messages WHERE %s = ? ORDER BY created_at ASC `, dialect.JSONExtract(drv, "metadata", "pending_id")) @@ -360,7 +363,7 @@ func (r *Repository) FindMessagesByPendingID(ctx context.Context, pendingID stri func (r *Repository) FindPendingClarificationMessagesBySessionID(ctx context.Context, sessionID string) ([]*models.Message, error) { drv := r.ro.DriverName() query := fmt.Sprintf(` - SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at + SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at FROM task_session_messages WHERE task_session_id = ? AND type = 'clarification_request' AND %s = 'pending' ORDER BY created_at ASC @@ -382,7 +385,7 @@ func (r *Repository) FindPendingClarificationMessagesBySessionID(ctx context.Con func (r *Repository) FindMessageByPendingIDAndQuestion(ctx context.Context, sessionID, pendingID, questionID string) (*models.Message, error) { drv := r.ro.DriverName() query := fmt.Sprintf(` - SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at + SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at FROM task_session_messages WHERE task_session_id = ? AND %s = ? AND %s = ? ORDER BY created_at ASC LIMIT 1 @@ -396,7 +399,7 @@ func (r *Repository) FindMessageByPendingIDAndQuestion(ctx context.Context, sess var metadataJSON string err := r.ro.QueryRowContext(ctx, r.ro.Rebind(query), sessionID, pendingID, questionID).Scan( &message.ID, &message.TaskSessionID, &message.TaskID, &message.TurnID, &message.AuthorType, &message.AuthorID, - &message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt, + &message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt, &message.UpdatedAt, ) if err != nil { return nil, err @@ -420,7 +423,7 @@ func (r *Repository) CompletePendingToolCallsForTurn(ctx context.Context, turnID drv := r.db.DriverName() query := fmt.Sprintf(` UPDATE task_session_messages - SET metadata = %s + SET metadata = %s, updated_at = CURRENT_TIMESTAMP WHERE turn_id = ? AND type != 'permission_request' AND %s NOT IN ('complete', 'error') @@ -449,10 +452,11 @@ func (r *Repository) UpdateMessage(ctx context.Context, message *models.Message) requestsInput = 1 } + message.UpdatedAt = time.Now().UTC() result, err := r.db.ExecContext(ctx, r.db.Rebind(` - UPDATE task_session_messages SET content = ?, requests_input = ?, type = ?, metadata = ? + UPDATE task_session_messages SET content = ?, requests_input = ?, type = ?, metadata = ?, updated_at = ? WHERE id = ? - `), message.Content, requestsInput, string(message.Type), string(metadataJSON), message.ID) + `), message.Content, requestsInput, string(message.Type), string(metadataJSON), message.UpdatedAt, message.ID) if err != nil { return err } diff --git a/apps/backend/internal/task/repository/sqlite/message_test.go b/apps/backend/internal/task/repository/sqlite/message_test.go index 1fd21e5863..7c09220173 100644 --- a/apps/backend/internal/task/repository/sqlite/message_test.go +++ b/apps/backend/internal/task/repository/sqlite/message_test.go @@ -4,6 +4,8 @@ import ( "context" "testing" "time" + + "github.com/kandev/kandev/internal/task/models" ) // insertMsgWithType inserts a message row with a configurable type column, @@ -58,6 +60,48 @@ func TestListMessagesByTurnID(t *testing.T) { } } +func TestUpdateMessageBumpsUpdatedAt(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + seedForMsgTest(t, repo, "task-U", "sess-U", "turn-U") + + created := time.Now().UTC().Add(-time.Hour) + msg := &models.Message{ + ID: "m-u", + TaskSessionID: "sess-U", + TurnID: "turn-U", + AuthorType: models.MessageAuthorAgent, + Content: "hello", + Type: models.MessageTypeMessage, + CreatedAt: created, + } + if err := repo.CreateMessage(ctx, msg); err != nil { + t.Fatalf("CreateMessage: %v", err) + } + + // On insert, updated_at defaults to created_at. + got, err := repo.GetMessage(ctx, "m-u") + if err != nil { + t.Fatalf("GetMessage: %v", err) + } + if !got.UpdatedAt.Equal(got.CreatedAt) { + t.Errorf("after create, updated_at = %v, want == created_at %v", got.UpdatedAt, got.CreatedAt) + } + + // Update advances updated_at past created_at. + msg.Content = "hello world" + if err := repo.UpdateMessage(ctx, msg); err != nil { + t.Fatalf("UpdateMessage: %v", err) + } + got, err = repo.GetMessage(ctx, "m-u") + if err != nil { + t.Fatalf("GetMessage after update: %v", err) + } + if !got.UpdatedAt.After(got.CreatedAt) { + t.Errorf("after update, updated_at = %v, want after created_at %v", got.UpdatedAt, got.CreatedAt) + } +} + func TestCountToolCallMessagesBySession_Empty(t *testing.T) { repo := newRepoForSessionTests(t) got, err := repo.CountToolCallMessagesBySession(context.Background(), nil) diff --git a/apps/backend/internal/task/service/service_events.go b/apps/backend/internal/task/service/service_events.go index d2e50ba5d1..d706d90d87 100644 --- a/apps/backend/internal/task/service/service_events.go +++ b/apps/backend/internal/task/service/service_events.go @@ -324,7 +324,11 @@ func (s *Service) publishMessageEvent(ctx context.Context, eventType string, mes "content": sysprompt.StripSystemContent(message.Content), "type": messageType, "requests_input": message.RequestsInput, - "created_at": message.CreatedAt.Format(time.RFC3339), + // RFC3339Nano keeps sub-second precision so rapid updates within the same + // second produce distinct timestamps; the REST/DTO path serializes these + // fields with nanosecond precision too, so both delivery channels agree. + "created_at": message.CreatedAt.Format(time.RFC3339Nano), + "updated_at": message.UpdatedAt.Format(time.RFC3339Nano), } if hasHidden { diff --git a/apps/backend/pkg/api/v1/task.go b/apps/backend/pkg/api/v1/task.go index 1479ef9669..6d1816de23 100644 --- a/apps/backend/pkg/api/v1/task.go +++ b/apps/backend/pkg/api/v1/task.go @@ -164,6 +164,7 @@ type Message struct { RequestsInput bool `json:"requests_input"` // True if agent is requesting user input Metadata map[string]interface{} `json:"metadata,omitempty"` CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at,omitempty"` // Authoritative per-message change signal } // CreateMessageRequest for adding a message to a task session diff --git a/apps/cli/src/process.test.ts b/apps/cli/src/process.test.ts new file mode 100644 index 0000000000..3e50a37f1b --- /dev/null +++ b/apps/cli/src/process.test.ts @@ -0,0 +1,63 @@ +import { afterAll, describe, expect, it } from "vitest"; + +import { __resetBrokenPipeGuard, ignoreBrokenPipe } from "./process"; + +function makeError(code: string, message: string): NodeJS.ErrnoException { + return Object.assign(new Error(message), { code }); +} + +function lastErrorListener(stream: NodeJS.WriteStream): (err: NodeJS.ErrnoException) => void { + const listeners = stream.listeners("error"); + return listeners[listeners.length - 1] as (err: NodeJS.ErrnoException) => void; +} + +describe("ignoreBrokenPipe", () => { + // Capture the handler we install so we can both exercise and clean it up. + let installed: ((err: NodeJS.ErrnoException) => void) | undefined; + + afterAll(() => { + if (installed) { + process.stdout.removeListener("error", installed); + process.stderr.removeListener("error", installed); + } + // Reset the module guard alongside listener removal so a later + // ignoreBrokenPipe() reattaches instead of hitting a no-op with no + // listeners present (order-independent for other tests in this worker). + __resetBrokenPipeGuard(); + }); + + it("attaches an error handler to stdout and stderr", () => { + const stdoutBefore = process.stdout.listeners("error").length; + const stderrBefore = process.stderr.listeners("error").length; + + ignoreBrokenPipe(); + + expect(process.stdout.listeners("error").length).toBe(stdoutBefore + 1); + expect(process.stderr.listeners("error").length).toBe(stderrBefore + 1); + + // stdout and stderr share the same handler reference. + installed = lastErrorListener(process.stdout); + expect(lastErrorListener(process.stderr)).toBe(installed); + }); + + it("swallows EPIPE errors without throwing", () => { + const handler = lastErrorListener(process.stdout); + expect(() => handler(makeError("EPIPE", "write EPIPE"))).not.toThrow(); + }); + + it("rethrows non-EPIPE errors", () => { + const handler = lastErrorListener(process.stdout); + expect(() => handler(makeError("EACCES", "boom"))).toThrow("boom"); + }); + + it("is idempotent across repeated calls", () => { + const stdoutCount = process.stdout.listeners("error").length; + const stderrCount = process.stderr.listeners("error").length; + + ignoreBrokenPipe(); + ignoreBrokenPipe(); + + expect(process.stdout.listeners("error").length).toBe(stdoutCount); + expect(process.stderr.listeners("error").length).toBe(stderrCount); + }); +}); diff --git a/apps/cli/src/process.ts b/apps/cli/src/process.ts index 0c8c5f6094..6d6d460506 100644 --- a/apps/cli/src/process.ts +++ b/apps/cli/src/process.ts @@ -36,6 +36,12 @@ export function createProcessSupervisor(): { }; const attachSignalHandlers = () => { + // Ctrl-C sends SIGINT to the whole process group, so the parent (make/shell) + // exits and closes the pipe our stdout/stderr write to. Any console.log during + // shutdown then triggers an EPIPE 'error' event with no listener, which Node + // treats as fatal and crashes us before children are gracefully terminated. + // Swallow EPIPE so shutdown can finish. + ignoreBrokenPipe(); process.on("SIGINT", onSignal); // SIGTERM is not available on Windows — only attach where supported if (process.platform !== "win32") { @@ -46,6 +52,32 @@ export function createProcessSupervisor(): { return { children, shutdown, attachSignalHandlers }; } +let brokenPipeGuarded = false; + +/** + * Install error handlers on stdout/stderr that swallow EPIPE (broken pipe). + * Idempotent: only attaches once per process. + */ +export function ignoreBrokenPipe(): void { + if (brokenPipeGuarded) return; + brokenPipeGuarded = true; + const onPipeError = (err: NodeJS.ErrnoException) => { + if (err.code === "EPIPE") return; + throw err; + }; + process.stdout.on("error", onPipeError); + process.stderr.on("error", onPipeError); +} + +/** + * Test-only: resets the broken-pipe guard so a later `ignoreBrokenPipe()` call + * reattaches listeners. Keeps the module guard consistent with suites that + * remove the listeners they installed. + */ +export function __resetBrokenPipeGuard(): void { + brokenPipeGuarded = false; +} + /** * Terminate a process and wait for it to exit. * On Unix: sends SIGTERM, falls back to SIGKILL after timeout. diff --git a/apps/web/app/globals.css b/apps/web/app/globals.css index d5a1f9dbe9..84a54c5914 100644 --- a/apps/web/app/globals.css +++ b/apps/web/app/globals.css @@ -360,10 +360,10 @@ @keyframes border-flash-glow { 0%, 100% { - box-shadow: inset 0 0 0 1px color-mix(in oklab, var(--primary) 20%, transparent); + opacity: 0.35; } 50% { - box-shadow: inset 0 0 0 1px color-mix(in oklab, var(--primary) 60%, transparent); + opacity: 1; } } @@ -410,71 +410,72 @@ border-radius: 4px; } -@keyframes chat-input-running { - 0%, - 100% { - border-color: color-mix(in oklab, var(--primary) 40%, var(--border)); - box-shadow: - 0 0 6px 1px color-mix(in oklab, var(--primary) 12%, transparent), - 0 0 15px 3px color-mix(in oklab, var(--primary) 6%, transparent); - } - 50% { - border-color: color-mix(in oklab, var(--primary) 70%, var(--border)); - box-shadow: - 0 0 20px 4px color-mix(in oklab, var(--primary) 20%, transparent), - 0 0 50px 8px color-mix(in oklab, var(--primary) 10%, transparent); - } -} - -.chat-input-running { - animation: chat-input-running 3s ease-in-out infinite; -} - -@keyframes chat-input-running-plan { - 0%, - 100% { - border-color: color-mix(in oklab, var(--primary) 40%, var(--border)); - box-shadow: - 0 0 6px 1px color-mix(in oklab, var(--primary) 12%, transparent), - 0 0 15px 3px color-mix(in oklab, var(--primary) 6%, transparent); - } - 50% { - border-color: color-mix(in oklab, var(--primary) 70%, var(--border)); - box-shadow: - 0 0 20px 4px color-mix(in oklab, var(--primary) 20%, transparent), - 0 0 50px 8px color-mix(in oklab, var(--primary) 10%, transparent); - } -} - +/* Chat-input "running" affordance. + * + * The colored border is a static border-color (painted once). The pulsing glow + * lives on a pseudo-element that animates only `opacity`, which the compositor + * handles on the GPU without a per-frame main-thread paint. The glow is driven + * from the OUTER `.relative` wrapper (`.chat-input-glow*`) because the inner + * input box has `overflow-hidden`, which would clip a child's outer box-shadow. */ +.chat-input-running, .chat-input-running-plan { - animation: chat-input-running-plan 3s ease-in-out infinite; -} - -@keyframes chat-input-starting { - 0%, - 100% { - border-color: color-mix(in oklab, var(--primary) 40%, var(--border)); - box-shadow: - 0 0 4px 1px color-mix(in oklab, var(--primary) 10%, transparent), - 0 0 10px 2px color-mix(in oklab, var(--primary) 5%, transparent); - } - 50% { - border-color: color-mix(in oklab, var(--primary) 60%, var(--border)); - box-shadow: - 0 0 12px 3px color-mix(in oklab, var(--primary) 15%, transparent), - 0 0 30px 6px color-mix(in oklab, var(--primary) 8%, transparent); - } + border-color: color-mix(in oklab, var(--primary) 55%, var(--border)); } .chat-input-starting { - animation: chat-input-starting 2s ease-in-out infinite; + border-color: color-mix(in oklab, var(--primary) 50%, var(--border)); +} + +/* The glow pseudo-element sits at `z-index: -1` so the input content paints on + * top of it. `isolation: isolate` makes the wrapper its own stacking context so + * that negative z-index stays local instead of dropping behind ancestor + * backgrounds (which would hide the glow entirely). */ +.chat-input-glow-running, +.chat-input-glow-starting { + position: relative; + isolation: isolate; +} + +@keyframes chat-input-glow-pulse { + 0%, + 100% { + opacity: 0.4; + } + 50% { + opacity: 1; + } +} + +.chat-input-glow-running::after, +.chat-input-glow-starting::after { + content: ""; + position: absolute; + inset: 0; + border-radius: 0.25rem; + pointer-events: none; + z-index: -1; + animation: chat-input-glow-pulse 3s ease-in-out infinite; +} + +.chat-input-glow-running::after { + box-shadow: + 0 0 20px 4px color-mix(in oklab, var(--primary) 20%, transparent), + 0 0 50px 8px color-mix(in oklab, var(--primary) 10%, transparent); +} + +.chat-input-glow-starting::after { + box-shadow: + 0 0 12px 3px color-mix(in oklab, var(--primary) 15%, transparent), + 0 0 30px 6px color-mix(in oklab, var(--primary) 8%, transparent); + animation-duration: 2s; } @media (prefers-reduced-motion: reduce) { .animate-changes-flash, - .chat-input-running, - .chat-input-running-plan, - .chat-input-starting, + .chat-input-glow-running::after, + .chat-input-glow-starting::after, + .border-flash-animation::after, + .node-border-running::after, .search-flash, .animate-queue-open, .animate-queue-close { @@ -494,6 +495,7 @@ border-radius: inherit; pointer-events: none; z-index: 10; + box-shadow: inset 0 0 0 1px color-mix(in oklab, var(--primary) 60%, transparent); animation: border-flash-glow 3s ease-in-out infinite; } } @@ -1392,24 +1394,34 @@ border-radius: 8px; } -/* Pipeline node rotating border for running tasks */ -@keyframes node-border-pulse { +/* Pipeline node glow for running tasks — static ring on the element, the pulse + * is opacity-only on a pseudo-element so it composites on the GPU rather than + * repainting the box-shadow every frame. */ +@keyframes node-border-glow { 0%, 100% { - box-shadow: - 0 0 0 1px color-mix(in oklab, var(--primary) 42%, var(--border)), - 0 0 0 0 color-mix(in oklab, var(--primary) 0%, transparent); + opacity: 0.5; } 50% { - box-shadow: - 0 0 0 1px color-mix(in oklab, var(--primary) 68%, var(--border)), - 0 0 0 3px color-mix(in oklab, var(--primary) 12%, transparent); + opacity: 1; } } .node-border-running { position: relative; - animation: node-border-pulse 2.8s ease-in-out infinite; + box-shadow: 0 0 0 1px color-mix(in oklab, var(--primary) 42%, var(--border)); +} + +.node-border-running::after { + content: ""; + position: absolute; + inset: 0; + border-radius: inherit; + pointer-events: none; + box-shadow: + 0 0 0 1px color-mix(in oklab, var(--primary) 68%, var(--border)), + 0 0 0 3px color-mix(in oklab, var(--primary) 12%, transparent); + animation: node-border-glow 2.8s ease-in-out infinite; } .chat-message-list { diff --git a/apps/web/app/office/tasks/[id]/advanced-panels/chat-panel.tsx b/apps/web/app/office/tasks/[id]/advanced-panels/chat-panel.tsx index b16c0a5c86..6471b20773 100644 --- a/apps/web/app/office/tasks/[id]/advanced-panels/chat-panel.tsx +++ b/apps/web/app/office/tasks/[id]/advanced-panels/chat-panel.tsx @@ -12,7 +12,7 @@ import { useAppStore } from "@/components/state-provider"; import { getWebSocketClient } from "@/lib/ws/connection"; import { buildStartRequest } from "@/lib/services/session-launch-helpers"; import { MessageRenderer } from "@/components/task/chat/message-renderer"; -import type { Message, TaskSessionState } from "@/lib/types/http"; +import type { Message } from "@/lib/types/http"; type AdvancedChatPanelProps = { taskId: string; @@ -58,14 +58,12 @@ function StartSessionPrompt({ function MessageList({ messages, isLoading, - sessionState, taskId, sessionId, scrollRef, }: { messages: Message[]; isLoading: boolean; - sessionState: TaskSessionState | null; taskId: string; sessionId: string | null; scrollRef: React.RefObject; @@ -87,7 +85,6 @@ function MessageList({ key={msg.id} comment={msg} isTaskDescription={idx === 0 && msg.author_type === "user"} - sessionState={sessionState ?? undefined} taskId={taskId} sessionId={sessionId ?? undefined} /> @@ -198,7 +195,6 @@ export function AdvancedChatPanel({ taskId, sessionId, hideInput }: AdvancedChat diff --git a/apps/web/components/shared/markdown-components.tsx b/apps/web/components/shared/markdown-components.tsx index 108d7a7da2..c912a7836b 100644 --- a/apps/web/components/shared/markdown-components.tsx +++ b/apps/web/components/shared/markdown-components.tsx @@ -13,69 +13,10 @@ import { isMermaidContent } from "@/components/editors/tiptap/tiptap-mermaid-ext // eslint-disable-next-line @typescript-eslint/no-explicit-any export const remarkPlugins: any[] = [remarkGfm, remarkBreaks, remarkGemoji]; -const FENCE_OPEN_RE = /^ {0,3}(`{3,})/; -const TRAILING_FENCE_RE = /(`{3,})\s*$/; - -function pureCloseLength(line: string, openCount: number): number | null { - const match = /^ {0,3}(`{3,})\s*$/.exec(line); - if (!match || match[1].length < openCount) return null; - return match[1].length; -} - -function gluedCloseLength(line: string, openCount: number): number | null { - const match = TRAILING_FENCE_RE.exec(line); - if (!match || match[1].length < openCount) return null; - // Reject pure-fence lines (already handled by pureCloseLength) and lines - // where everything before the trailing run is whitespace only. - const head = line.slice(0, line.length - match[0].length).trimEnd(); - if (head.length === 0) return null; - return match[1].length; -} - -/** - * Pre-process a markdown string to repair fenced code blocks that have their - * closing fence glued to the last code line (`...}\`\`\`\n`prose`). Without - * this, CommonMark/GFM treats the glued backticks as code content, so the - * fence never closes and following prose gets swallowed into one huge code - * node. We split such lines into `\n` only when we're - * inside an open fence whose opener run length is ≤ the trailing run length. - * - * Pure string preprocessing, intentionally not a remark plugin. - */ -export function normalizeMarkdown(input: string): string { - if (!input || input.length === 0) return input; - const hadTrailingNewline = input.endsWith("\n"); - const lines = input.split("\n"); - const out: string[] = []; - let openCount: number | null = null; - - for (const line of lines) { - if (openCount === null) { - const opener = FENCE_OPEN_RE.exec(line); - if (opener) openCount = opener[1].length; - out.push(line); - continue; - } - if (pureCloseLength(line, openCount) !== null) { - openCount = null; - out.push(line); - continue; - } - const glued = gluedCloseLength(line, openCount); - if (glued !== null) { - const trailingMatch = TRAILING_FENCE_RE.exec(line)!; - const head = line.slice(0, line.length - trailingMatch[0].length); - out.push(head); - out.push("`".repeat(glued)); - openCount = null; - continue; - } - out.push(line); - } - - const result = out.join("\n"); - return hadTrailingNewline && !result.endsWith("\n") ? result + "\n" : result; -} +// `normalizeMarkdown` (pure string transform) and its cached variant live in +// the React-free markdown cache module. Re-exported here so existing importers +// keep working. +export { normalizeMarkdown } from "@/lib/markdown/normalize-cache"; /** * Recursively extracts text content from React children. diff --git a/apps/web/components/shared/memoized-markdown.test.tsx b/apps/web/components/shared/memoized-markdown.test.tsx new file mode 100644 index 0000000000..1b25558199 --- /dev/null +++ b/apps/web/components/shared/memoized-markdown.test.tsx @@ -0,0 +1,53 @@ +import { useState } from "react"; +import { fireEvent, render } from "@testing-library/react"; +import { afterEach, describe, expect, it, vi } from "vitest"; + +const markdownSpy = vi.fn(); + +vi.mock("react-markdown", () => ({ + default: ({ children }: { children: string }) => { + markdownSpy(children); + return
{children}
; + }, +})); + +// Stub the component overrides so the test stays light (no shiki / mermaid). +vi.mock("@/components/shared/markdown-components", () => ({ + markdownComponents: {}, + remarkPlugins: [], +})); + +import { MemoizedMarkdown } from "./memoized-markdown"; +import { __resetMarkdownCounters } from "@/lib/markdown/normalize-cache"; + +function Parent({ content }: { content: string }) { + const [tick, setTick] = useState(0); + return ( +
+ + +
+ ); +} + +describe("MemoizedMarkdown", () => { + afterEach(() => { + markdownSpy.mockClear(); + __resetMarkdownCounters(); + }); + + it("does not re-render markdown when the parent re-renders with same content", () => { + const { getByText } = render(); + expect(markdownSpy).toHaveBeenCalledTimes(1); + fireEvent.click(getByText("tick")); + fireEvent.click(getByText("tick")); + expect(markdownSpy).toHaveBeenCalledTimes(1); + }); + + it("re-renders markdown when content changes", () => { + const { rerender } = render(); + expect(markdownSpy).toHaveBeenCalledTimes(1); + rerender(); + expect(markdownSpy).toHaveBeenCalledTimes(2); + }); +}); diff --git a/apps/web/components/shared/memoized-markdown.tsx b/apps/web/components/shared/memoized-markdown.tsx new file mode 100644 index 0000000000..3aedf0002c --- /dev/null +++ b/apps/web/components/shared/memoized-markdown.tsx @@ -0,0 +1,23 @@ +"use client"; + +import { memo } from "react"; +import ReactMarkdown from "react-markdown"; +import { markdownComponents, remarkPlugins } from "@/components/shared/markdown-components"; +import { normalizeCached } from "@/lib/markdown/normalize-cache"; + +/** + * Markdown renderer behind a `memo` boundary keyed on the `content` string. + * + * `content` is a primitive, so React compares it by value. Object-ref churn + * from a message refetch can therefore never trigger a re-parse: an identical + * string bails out of the memo and re-uses the previously parsed element tree. + * The normalized string itself is also cached (LRU) so two messages with the + * same content share a single normalize pass. + */ +export const MemoizedMarkdown = memo(function MemoizedMarkdown({ content }: { content: string }) { + return ( + + {normalizeCached(content)} + + ); +}); diff --git a/apps/web/components/task/chat/__evals__/render-cost.test.tsx b/apps/web/components/task/chat/__evals__/render-cost.test.tsx new file mode 100644 index 0000000000..be0c1f8770 --- /dev/null +++ b/apps/web/components/task/chat/__evals__/render-cost.test.tsx @@ -0,0 +1,123 @@ +/** + * Render-cost eval harness — the objective guard against the markdown + * re-render/re-parse "storm". + * + * It mounts a list of memoized markdown rows and asserts how many real + * normalize passes (cache misses) and per-row re-renders happen under each + * update pattern. Wave 1 owns the parse-count assertions; Wave 2/3 extend the + * row-render assertions once callbacks + store identity are stabilized. + */ +import { memo, useState } from "react"; +import { act, fireEvent, render } from "@testing-library/react"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("react-markdown", () => ({ + default: ({ children }: { children: string }) =>
{children}
, +})); + +vi.mock("@/components/shared/markdown-components", () => ({ + markdownComponents: {}, + remarkPlugins: [], +})); + +import { MemoizedMarkdown } from "@/components/shared/memoized-markdown"; +import { __markdownParseCount, __resetMarkdownCounters } from "@/lib/markdown/normalize-cache"; + +type Row = { id: string; content: string }; + +const rowRenderCounts = new Map(); + +const MarkdownRow = memo(function MarkdownRow({ id, content }: Row) { + rowRenderCounts.set(id, (rowRenderCounts.get(id) ?? 0) + 1); + return ; +}); + +function MessageList({ rows, panel }: { rows: Row[]; panel: boolean }) { + return ( +
+ {rows.map((r) => ( + + ))} +
+ ); +} + +function makeRows(n: number): Row[] { + return Array.from({ length: n }, (_, i) => ({ id: `m${i}`, content: `# Message ${i}` })); +} + +/** Deep clone into fresh object refs but identical content (simulates a refetch). */ +function refetch(rows: Row[]): Row[] { + return rows.map((r) => ({ id: r.id, content: r.content })); +} + +function totalRowRenders(): number { + let total = 0; + for (const count of rowRenderCounts.values()) total += count; + return total; +} + +describe("chat render-cost evals", () => { + beforeEach(() => { + __resetMarkdownCounters(); + rowRenderCounts.clear(); + }); + + it("scenario 1: no-op refetch causes 0 parses and 0 row re-renders", () => { + const rows = makeRows(20); + const { rerender } = render(); + const parsesAfterMount = __markdownParseCount(); + rowRenderCounts.clear(); + + rerender(); + + expect(__markdownParseCount() - parsesAfterMount).toBe(0); + expect(totalRowRenders()).toBe(0); + }); + + it("scenario 2: one changed message causes exactly 1 new parse", () => { + const rows = makeRows(20); + const { rerender } = render(); + const parsesAfterMount = __markdownParseCount(); + rowRenderCounts.clear(); + + const next = refetch(rows); + next[5] = { id: next[5].id, content: "# Message 5 (edited)" }; + rerender(); + + expect(__markdownParseCount() - parsesAfterMount).toBe(1); + expect(rowRenderCounts.get("m5")).toBe(1); + }); + + it("scenario 3: unrelated parent re-render causes 0 parses", () => { + const rows = makeRows(20); + function Harness() { + const [panel, setPanel] = useState(false); + return ( +
+ + +
+ ); + } + const { getByText } = render(); + const parsesAfterMount = __markdownParseCount(); + rowRenderCounts.clear(); + + act(() => { + fireEvent.click(getByText("toggle")); + }); + + expect(__markdownParseCount() - parsesAfterMount).toBe(0); + expect(totalRowRenders()).toBe(0); + }); + + it("scenario 4: identical content across two rows parses once (cache hit)", () => { + const rows: Row[] = [ + { id: "a", content: "# Same content" }, + { id: "b", content: "# Same content" }, + ]; + render(); + expect(__markdownParseCount()).toBe(1); + }); +}); diff --git a/apps/web/components/task/chat/chat-input-body.tsx b/apps/web/components/task/chat/chat-input-body.tsx index 1141c038a4..d7c034c9e4 100644 --- a/apps/web/components/task/chat/chat-input-body.tsx +++ b/apps/web/components/task/chat/chat-input-body.tsx @@ -233,6 +233,15 @@ export type ChatInputBodyProps = { editorAreaProps: ChatInputEditorAreaProps; }; +/** Glow class for the outer wrapper. The pulsing glow lives on the wrapper + * (not the inner box) because the inner box has `overflow-hidden`, which would + * clip a child pseudo-element's outer box-shadow. */ +function chatInputGlowClass(isAgentBusy: boolean, isStarting: boolean): string { + if (isAgentBusy) return "chat-input-glow-running"; + if (isStarting) return "chat-input-glow-starting"; + return ""; +} + export function ChatInputBody({ containerRef, height, @@ -294,7 +303,7 @@ export function ChatInputBody({ ); return ( -
+
@@ -256,12 +252,7 @@ export const NativeMessageList = memo(function NativeMessageList({ {(footerActionMessages ?? []).map((msg: Message) => ( - + ))} {/* Bottom anchor — browser keeps scroll pinned here when new content appends */} diff --git a/apps/web/components/task/chat/message-list-shared.test.tsx b/apps/web/components/task/chat/message-list-shared.test.tsx new file mode 100644 index 0000000000..2a56024c99 --- /dev/null +++ b/apps/web/components/task/chat/message-list-shared.test.tsx @@ -0,0 +1,75 @@ +import { useState } from "react"; +import { fireEvent, render } from "@testing-library/react"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { RenderItem } from "@/hooks/use-processed-messages"; +import type { Message } from "@/lib/types/http"; + +const rendererSpy = vi.fn(); + +vi.mock("@/components/task/chat/message-renderer", () => ({ + MessageRenderer: (props: { onOpenFile?: unknown }) => { + rendererSpy(props); + return
; + }, +})); +vi.mock("@/components/task/chat/messages/turn-group-message", () => ({ + TurnGroupMessage: () =>
, +})); +vi.mock("@/components/session/prepare-progress", () => ({ + PrepareProgress: () =>
, +})); + +import { MessageItem } from "./message-list-shared"; + +const item: RenderItem = { type: "message", message: { id: "m1" } as Message }; +const noop = () => {}; +const perm = new Map(); +const kids = new Map(); + +function row(onOpenFile: (p: string) => void) { + return ( + + ); +} + +function Harness({ onOpenFile }: { onOpenFile: (p: string) => void }) { + const [, setTick] = useState(0); + return ( +
+ + {row(onOpenFile)} +
+ ); +} + +describe("MessageItem memo boundary", () => { + afterEach(() => { + rendererSpy.mockClear(); + }); + + it("does not re-render the row when the parent re-renders with stable props", () => { + const { getByText } = render(); + expect(rendererSpy).toHaveBeenCalledTimes(1); + fireEvent.click(getByText("tick")); + fireEvent.click(getByText("tick")); + expect(rendererSpy).toHaveBeenCalledTimes(1); // memo bailed on stable props + }); + + it("re-renders the row when onOpenFile identity changes (stability requirement)", () => { + const { rerender } = render(row(() => {})); + expect(rendererSpy).toHaveBeenCalledTimes(1); + rerender(row(() => {})); + expect(rendererSpy).toHaveBeenCalledTimes(2); // fresh callback ref breaks memo + }); +}); diff --git a/apps/web/components/task/chat/message-list-shared.tsx b/apps/web/components/task/chat/message-list-shared.tsx index 6cf8a88a41..2cfc493943 100644 --- a/apps/web/components/task/chat/message-list-shared.tsx +++ b/apps/web/components/task/chat/message-list-shared.tsx @@ -1,8 +1,9 @@ "use client"; +import { memo } from "react"; import { Button } from "@kandev/ui/button"; import { GridSpinner } from "@/components/grid-spinner"; -import type { Message, TaskSessionState, TaskState } from "@/lib/types/http"; +import type { Message, TaskSessionState } from "@/lib/types/http"; import type { RenderItem } from "@/hooks/use-processed-messages"; import { MessageRenderer } from "@/components/task/chat/message-renderer"; import { TurnGroupMessage } from "@/components/task/chat/messages/turn-group-message"; @@ -20,7 +21,6 @@ export type MessageListProps = { messagesLoading: boolean; isWorking: boolean; sessionState?: TaskSessionState; - taskState?: TaskState; worktreePath?: string; onOpenFile?: (path: string) => void; }; @@ -100,7 +100,7 @@ export function MessageListStatus({ ); } -export function MessageItem({ +export const MessageItem = memo(function MessageItem({ item, sessionId, permissionsByToolCallId, @@ -110,9 +110,6 @@ export function MessageItem({ onOpenFile, isLastGroup, isTurnActive, - messages, - sessionState, - taskState, onScrollToMessage, }: { item: RenderItem; @@ -124,9 +121,6 @@ export function MessageItem({ onOpenFile?: (path: string) => void; isLastGroup: boolean; isTurnActive: boolean; - messages: Message[]; - sessionState?: TaskSessionState; - taskState?: TaskState; onScrollToMessage: (id: string) => void; }) { if (item.type === "prepare_progress") { @@ -144,9 +138,6 @@ export function MessageItem({ onOpenFile={onOpenFile} isLastGroup={isLastGroup} isTurnActive={isTurnActive} - allMessages={messages} - sessionState={sessionState} - taskState={taskState} onScrollToMessage={onScrollToMessage} /> ); @@ -155,16 +146,13 @@ export function MessageItem({ ); -} +}); diff --git a/apps/web/components/task/chat/message-list-virtuoso.tsx b/apps/web/components/task/chat/message-list-virtuoso.tsx index 16a9dffc31..b86e1cabb9 100644 --- a/apps/web/components/task/chat/message-list-virtuoso.tsx +++ b/apps/web/components/task/chat/message-list-virtuoso.tsx @@ -97,15 +97,7 @@ function useStableFirstItemIndex(items: RenderItem[]) { function useVirtuosoCallbacks(props: VirtuosoBodyProps) { const { items, sessionId, permissionsByToolCallId, childrenByParentToolCallId, taskId } = props; - const { - worktreePath, - onOpenFile, - lastTurnGroupId, - isRunning, - messages, - sessionState, - taskState, - } = props; + const { worktreePath, onOpenFile, lastTurnGroupId, isRunning } = props; const { hasMore, isLoadingMore, loadMore } = props; const virtuosoRef = useRef(null); const itemCount = items.length; @@ -162,9 +154,6 @@ function useVirtuosoCallbacks(props: VirtuosoBodyProps) { onOpenFile={onOpenFile} isLastGroup={item.type === "turn_group" && item.id === lastTurnGroupId} isTurnActive={isRunning} - messages={messages} - sessionState={sessionState} - taskState={taskState} onScrollToMessage={handleScrollToMessage} />
@@ -181,9 +170,6 @@ function useVirtuosoCallbacks(props: VirtuosoBodyProps) { onOpenFile, lastTurnGroupId, isRunning, - messages, - sessionState, - taskState, handleScrollToMessage, ], ); @@ -436,12 +422,7 @@ function useVirtuosoHeaderFooter(args: HeaderFooterArgs) { <> {footerActions.map((msg) => ( - + ))} ), @@ -511,12 +492,7 @@ export const VirtuosoMessageList = memo(function VirtuosoMessageList(props: Mess /> {footerActions.map((msg) => ( - + ))} ); diff --git a/apps/web/components/task/chat/message-renderer.tsx b/apps/web/components/task/chat/message-renderer.tsx index c675fee552..faa5a59d93 100644 --- a/apps/web/components/task/chat/message-renderer.tsx +++ b/apps/web/components/task/chat/message-renderer.tsx @@ -3,11 +3,12 @@ import { memo, useState, useCallback, type ReactElement } from "react"; import { IconPlayerPlay } from "@tabler/icons-react"; import { Button } from "@kandev/ui/button"; -import type { Message, TaskSessionState, TaskState } from "@/lib/types/http"; +import type { Message, TaskSessionState } from "@/lib/types/http"; import type { ToolCallMetadata } from "@/components/task/chat/types"; import { launchSession } from "@/lib/services/session-launch-service"; import { buildStartCreatedRequest } from "@/lib/services/session-launch-helpers"; import { useAppStore } from "@/components/state-provider"; +import { useTask } from "@/hooks/use-task"; import { ChatMessage } from "@/components/task/chat/messages/chat-message"; import { PermissionRequestMessage } from "@/components/task/chat/messages/permission-request-message"; import { StatusMessage } from "@/components/task/chat/messages/status-message"; @@ -31,15 +32,12 @@ import { type AdapterContext = { isTaskDescription: boolean; - sessionState?: TaskSessionState; - taskState?: TaskState; taskId?: string; permissionsByToolCallId?: Map; childrenByParentToolCallId?: Map; worktreePath?: string; sessionId?: string; onOpenFile?: (path: string) => void; - allMessages?: Message[]; onScrollToMessage?: (messageId: string) => void; }; @@ -81,6 +79,60 @@ function TaskDescriptionStartButton({ taskId, sessionId }: { taskId: string; ses ); } +function useSessionStateValue(sessionId?: string): TaskSessionState | undefined { + return useAppStore((state) => + sessionId ? (state.taskSessions.items[sessionId]?.state ?? undefined) : undefined, + ); +} + +/** + * The task-description message is the only row whose rendering depends on the + * live session/task state, so it reads them from the store directly. Keeping + * sessionState/taskState off the shared MessageRenderer props means a state + * transition no longer re-renders every message in the list. + */ +function TaskDescriptionMessage({ + comment, + taskId, + sessionId, + onScrollToMessage, +}: { + comment: Message; + taskId?: string; + sessionId?: string; + onScrollToMessage?: (messageId: string) => void; +}) { + const sessionState = useSessionStateValue(sessionId); + const task = useTask(taskId ?? null); + const renderAsUser = comment.author_type === "user" || sessionState !== "FAILED"; + if (!renderAsUser) { + return ( + + ); + } + const showStartButton = + sessionState === "CREATED" && task?.state !== "SCHEDULING" && !!taskId && !!sessionId; + return ( + <> + + {showStartButton && } + + ); +} + type MessageAdapter = { matches: (comment: Message, ctx: AdapterContext) => boolean; render: (comment: Message, ctx: AdapterContext) => ReactElement; @@ -220,7 +272,7 @@ const adapters: MessageAdapter[] = [ const meta = comment.metadata as Record | undefined; return Array.isArray(meta?.actions) && (meta.actions as unknown[]).length > 0; }, - render: (comment, ctx) => , + render: (comment) => , }, { matches: (comment) => @@ -247,29 +299,25 @@ const adapters: MessageAdapter[] = [ { matches: () => true, render: (comment, ctx) => { - if ( - comment.author_type === "user" || - (ctx.isTaskDescription && ctx.sessionState !== "FAILED") - ) { - const showStartButton = - ctx.isTaskDescription && - ctx.sessionState === "CREATED" && - ctx.taskState !== "SCHEDULING" && - ctx.taskId && - ctx.sessionId; + if (ctx.isTaskDescription) { return ( - <> - - {showStartButton && ( - - )} - + + ); + } + if (comment.author_type === "user") { + return ( + ); } return ( @@ -278,7 +326,7 @@ const adapters: MessageAdapter[] = [ label="Agent" className="bg-muted/40 text-foreground border-border/60" showRichBlocks={comment.type === "message" || comment.type === "content" || !comment.type} - allMessages={ctx.allMessages} + sessionId={ctx.sessionId} onScrollToMessage={ctx.onScrollToMessage} /> ); @@ -289,43 +337,34 @@ const adapters: MessageAdapter[] = [ type MessageRendererProps = { comment: Message; isTaskDescription: boolean; - sessionState?: TaskSessionState; - taskState?: TaskState; taskId?: string; permissionsByToolCallId?: Map; childrenByParentToolCallId?: Map; worktreePath?: string; sessionId?: string; onOpenFile?: (path: string) => void; - allMessages?: Message[]; onScrollToMessage?: (messageId: string) => void; }; export const MessageRenderer = memo(function MessageRenderer({ comment, isTaskDescription, - sessionState, - taskState, taskId, permissionsByToolCallId, childrenByParentToolCallId, worktreePath, sessionId, onOpenFile, - allMessages, onScrollToMessage, }: MessageRendererProps) { const ctx = { isTaskDescription, - sessionState, - taskState, taskId, permissionsByToolCallId, childrenByParentToolCallId, worktreePath, sessionId, onOpenFile, - allMessages, onScrollToMessage, }; const adapter = diff --git a/apps/web/components/task/chat/messages/action-message.test.tsx b/apps/web/components/task/chat/messages/action-message.test.tsx index 12c97827a7..1dcf3c9e04 100644 --- a/apps/web/components/task/chat/messages/action-message.test.tsx +++ b/apps/web/components/task/chat/messages/action-message.test.tsx @@ -1,9 +1,15 @@ import { afterEach, describe, expect, it, vi } from "vitest"; -import type { ReactNode } from "react"; import { cleanup, fireEvent, render, screen, waitFor } from "@testing-library/react"; import { StateProvider } from "@/components/state-provider"; import { ActionMessage } from "./action-message"; -import { sessionId as toSessionId, taskId as toTaskId, type Message } from "@/lib/types/http"; +import { + sessionId as toSessionId, + taskId as toTaskId, + type Message, + type TaskSession, + type TaskSessionState, +} from "@/lib/types/http"; +import type { AppState } from "@/lib/state/store"; const requestMock = vi.fn().mockResolvedValue({}); @@ -52,24 +58,29 @@ function retryMessage(overrides: Partial = {}): Message { } as Message; } -function Wrapper({ children }: { children: ReactNode }) { - return {children}; +/** ActionMessage reads session state from the store (keyed by comment.session_id), + * so seed it via the provider instead of passing a prop. */ +function renderAction(comment: Message, sessionState?: TaskSessionState) { + const initialState: Partial = sessionState + ? { taskSessions: { items: { "sess-1": { state: sessionState } as TaskSession } } } + : {}; + return render(, { + wrapper: ({ children }) => ( + {children} + ), + }); } describe("ActionMessage — transient retry (warning variant)", () => { it("renders the retrying copy in amber, not red", () => { - render(, { - wrapper: Wrapper, - }); + renderAction(retryMessage(), "WAITING_FOR_INPUT"); const text = screen.getByText(/retrying in 5s \(attempt 1\/3\)/i); expect(text.className).toContain("text-amber-600"); expect(text.className).not.toContain("text-red-600"); }); it("Cancel fires a session.recover ws_request with action cancel_retry", async () => { - render(, { - wrapper: Wrapper, - }); + renderAction(retryMessage(), "WAITING_FOR_INPUT"); fireEvent.click(screen.getByTestId(CANCEL_TEST_ID)); await waitFor(() => expect(requestMock).toHaveBeenCalledTimes(1)); expect(requestMock).toHaveBeenCalledWith("session.recover", { @@ -80,10 +91,7 @@ describe("ActionMessage — transient retry (warning variant)", () => { }); it("hides while the session is RUNNING (retry in flight) to avoid a stale card", () => { - const { container } = render( - , - { wrapper: Wrapper }, - ); + const { container } = renderAction(retryMessage(), "RUNNING"); expect(container.firstChild).toBeNull(); }); @@ -98,9 +106,7 @@ describe("ActionMessage — transient retry (warning variant)", () => { ], }, } as Partial); - render(, { - wrapper: Wrapper, - }); + renderAction(errorMsg, "WAITING_FOR_INPUT"); const text = screen.getByText(/Agent encountered an error/i); expect(text.className).toContain("text-red-600"); expect(text.className).not.toContain("text-amber-600"); diff --git a/apps/web/components/task/chat/messages/action-message.tsx b/apps/web/components/task/chat/messages/action-message.tsx index 09cbb7e631..b0fe5fff34 100644 --- a/apps/web/components/task/chat/messages/action-message.tsx +++ b/apps/web/components/task/chat/messages/action-message.tsx @@ -47,13 +47,15 @@ function isSessionActive(state?: TaskSessionState) { return state === "RUNNING" || state === "STARTING" || state === "COMPLETED"; } -export const ActionMessage = memo(function ActionMessage({ - comment, - sessionState, -}: { - comment: Message; - sessionState?: TaskSessionState; -}) { +export const ActionMessage = memo(function ActionMessage({ comment }: { comment: Message }) { + // Read session state from the store instead of receiving it as a prop, so a + // state transition doesn't re-render every message in the list (only the + // rare action messages that actually depend on it). + const sessionState = useAppStore((state) => + comment.session_id + ? (state.taskSessions.items[comment.session_id]?.state ?? undefined) + : undefined, + ); const metadata = comment.metadata as ActionMeta | undefined; const isWarning = metadata?.variant === "warning"; const message = comment.content || "An error occurred"; diff --git a/apps/web/components/task/chat/messages/agent-plan-message.tsx b/apps/web/components/task/chat/messages/agent-plan-message.tsx index 2b03ad7e9d..e7cd388f01 100644 --- a/apps/web/components/task/chat/messages/agent-plan-message.tsx +++ b/apps/web/components/task/chat/messages/agent-plan-message.tsx @@ -1,15 +1,10 @@ "use client"; import { useState, memo } from "react"; -import ReactMarkdown from "react-markdown"; import { IconMinus, IconPlus, IconMaximize } from "@tabler/icons-react"; import { Button } from "@kandev/ui/button"; import { Dialog, DialogContent, DialogHeader, DialogTitle } from "@kandev/ui/dialog"; -import { - markdownComponents, - normalizeMarkdown, - remarkPlugins, -} from "@/components/shared/markdown-components"; +import { MemoizedMarkdown } from "@/components/shared/memoized-markdown"; import type { Message } from "@/lib/types/http"; function parsePlanContent(text: string): { title: string; body: string } { @@ -28,9 +23,7 @@ function PlanMarkdownBody({ text, className }: { text: string; className?: strin
*]:my-2 [&>p]:my-2 [&>ul]:my-2 [&>ol]:my-2 ${className ?? ""}`} > - - {normalizeMarkdown(text)} - +
); } diff --git a/apps/web/components/task/chat/messages/chat-message.tsx b/apps/web/components/task/chat/messages/chat-message.tsx index 3cee6cf393..bbadc6d55d 100644 --- a/apps/web/components/task/chat/messages/chat-message.tsx +++ b/apps/web/components/task/chat/messages/chat-message.tsx @@ -1,19 +1,14 @@ "use client"; import { memo, useState, useCallback } from "react"; -import ReactMarkdown from "react-markdown"; import { IconWand, IconMessageDots, IconFile } from "@tabler/icons-react"; import { cn } from "@/lib/utils"; import type { Message } from "@/lib/types/http"; import { RichBlocks } from "@/components/task/chat/messages/rich-blocks"; import { MessageActions } from "@/components/task/chat/messages/message-actions"; -import { useMessageNavigation } from "@/hooks/use-message-navigation"; +import { useUserMessageNavigation } from "@/hooks/use-message-navigation"; import { SenderTaskBadge, type SenderTaskInfo } from "./sender-task-badge"; -import { - markdownComponents, - normalizeMarkdown, - remarkPlugins, -} from "@/components/shared/markdown-components"; +import { MemoizedMarkdown } from "@/components/shared/memoized-markdown"; import { ImagePreviewDialog } from "@/components/task/chat/image-preview-dialog"; import { WorkflowStepMessageBadge, @@ -27,7 +22,7 @@ type ChatMessageProps = { label: string; className: string; showRichBlocks?: boolean; - allMessages?: Message[]; + sessionId?: string | null; onScrollToMessage?: (messageId: string) => void; }; @@ -88,9 +83,7 @@ function renderUserMessageBody( if (hasContent) { return (
- - {normalizeMarkdown(content)} - +
); } @@ -106,7 +99,7 @@ type UserMessageProps = { comment: Message; showRaw: boolean; onToggleRaw: () => void; - allMessages?: Message[]; + sessionId?: string | null; onScrollToMessage?: (messageId: string) => void; }; @@ -200,10 +193,10 @@ function UserMessageContent({ comment, showRaw, onToggleRaw, - allMessages, + sessionId, onScrollToMessage, }: UserMessageProps) { - const userNavigation = useMessageNavigation(allMessages || [], comment.id, "user"); + const userNavigation = useUserMessageNavigation(sessionId ?? null, comment.id); const { imageAttachments, fileAttachments, @@ -263,15 +256,16 @@ function UserMessageContent({ showTimestamp={true} showRawToggle={true} hasHiddenPrompts={hasHiddenPrompts} - showNavigation={!!allMessages && allMessages.length > 0} + showNavigation={userNavigation.hasPrevious || userNavigation.hasNext} isRawView={showRaw} onToggleRaw={onToggleRaw} onNavigatePrev={() => { - if (userNavigation.previous && onScrollToMessage) - onScrollToMessage(userNavigation.previous.id); + if (userNavigation.previousId && onScrollToMessage) + onScrollToMessage(userNavigation.previousId); }} onNavigateNext={() => { - if (userNavigation.next && onScrollToMessage) onScrollToMessage(userNavigation.next.id); + if (userNavigation.nextId && onScrollToMessage) + onScrollToMessage(userNavigation.nextId); }} hasPrev={userNavigation.hasPrevious} hasNext={userNavigation.hasNext} @@ -300,9 +294,7 @@ function AgentMessageContent({ comment, showRaw, onToggleRaw, showRichBlocks }: ) : (
- - {normalizeMarkdown(comment.content || "(empty)")} - + {showRichBlocks ? : null}
)} @@ -328,7 +320,7 @@ export const ChatMessage = memo(function ChatMessage({ label, className, showRichBlocks, - allMessages, + sessionId, onScrollToMessage, }: ChatMessageProps) { const [showRaw, setShowRaw] = useState(false); @@ -360,7 +352,7 @@ export const ChatMessage = memo(function ChatMessage({ comment={comment} showRaw={showRaw} onToggleRaw={toggleRaw} - allMessages={allMessages} + sessionId={sessionId} onScrollToMessage={onScrollToMessage} /> ); diff --git a/apps/web/components/task/chat/messages/turn-group-message.tsx b/apps/web/components/task/chat/messages/turn-group-message.tsx index 01e7baad15..db8bfb8015 100644 --- a/apps/web/components/task/chat/messages/turn-group-message.tsx +++ b/apps/web/components/task/chat/messages/turn-group-message.tsx @@ -4,7 +4,7 @@ import { useState, useCallback, memo, useMemo } from "react"; import { IconChevronDown, IconChevronRight } from "@tabler/icons-react"; import { GridSpinner } from "@/components/grid-spinner"; import { cn, transformPathsInText } from "@/lib/utils"; -import type { Message, TaskSessionState, TaskState } from "@/lib/types/http"; +import type { Message } from "@/lib/types/http"; import type { TurnGroup } from "@/hooks/use-processed-messages"; import type { ToolCallMetadata } from "@/components/task/chat/types"; import { MessageRenderer } from "@/components/task/chat/message-renderer"; @@ -21,9 +21,6 @@ type TurnGroupMessageProps = { isLastGroup?: boolean; /** Whether the turn is still active (agent is running) */ isTurnActive?: boolean; - allMessages?: Message[]; - sessionState?: TaskSessionState; - taskState?: TaskState; onScrollToMessage?: (messageId: string) => void; }; @@ -154,9 +151,6 @@ type TurnGroupContentProps = { taskId?: string; worktreePath?: string; onOpenFile?: (path: string) => void; - allMessages?: Message[]; - sessionState?: TaskSessionState; - taskState?: TaskState; onScrollToMessage?: (messageId: string) => void; }; @@ -261,14 +255,11 @@ function renderMessageEntry(message: Message, props: MessageRenderProps) { comment={message} isTaskDescription={false} taskId={props.taskId} - sessionState={props.sessionState} - taskState={props.taskState} permissionsByToolCallId={props.permissionsByToolCallId} childrenByParentToolCallId={props.childrenByParentToolCallId} worktreePath={props.worktreePath} sessionId={props.sessionId ?? undefined} onOpenFile={props.onOpenFile} - allMessages={props.allMessages} onScrollToMessage={props.onScrollToMessage} /> ); @@ -320,9 +311,6 @@ function TurnGroupContent({ taskId, worktreePath, onOpenFile, - allMessages, - sessionState, - taskState, onScrollToMessage, }: TurnGroupContentProps) { const renderProps: MessageRenderProps = { @@ -332,9 +320,6 @@ function TurnGroupContent({ taskId, worktreePath, onOpenFile, - allMessages, - sessionState, - taskState, onScrollToMessage, }; const compacted = useMemo(() => compactTurnGroupMessages(group.messages), [group.messages]); @@ -361,9 +346,6 @@ export const TurnGroupMessage = memo(function TurnGroupMessage({ onOpenFile, isLastGroup = false, isTurnActive = false, - allMessages, - sessionState, - taskState, onScrollToMessage, }: TurnGroupMessageProps) { const isGroupRunning = hasRunningTool(group.messages); @@ -401,9 +383,6 @@ export const TurnGroupMessage = memo(function TurnGroupMessage({ taskId={taskId} worktreePath={worktreePath} onOpenFile={onOpenFile} - allMessages={allMessages} - sessionState={sessionState} - taskState={taskState} onScrollToMessage={onScrollToMessage} /> )} diff --git a/apps/web/components/task/chat/tiptap-input.tsx b/apps/web/components/task/chat/tiptap-input.tsx index 825fdb20b3..77878cd0cf 100644 --- a/apps/web/components/task/chat/tiptap-input.tsx +++ b/apps/web/components/task/chat/tiptap-input.tsx @@ -10,6 +10,7 @@ import { useMemo, } from "react"; import { EditorContent } from "@tiptap/react"; +import { useShallow } from "zustand/react/shallow"; import { useCustomPrompts } from "@/hooks/domains/settings/use-custom-prompts"; import { useAppStore, useAppStoreApi } from "@/components/state-provider"; import { getWebSocketClient } from "@/lib/ws/connection"; @@ -490,8 +491,15 @@ function TipTapPopups({ } function useMessageHistoryForSession(sessionId: string | null): string[] { - const messages = useAppStore((s) => (sessionId ? s.messages.bySession[sessionId] : undefined)); - return useMemo(() => extractUserHistory(messages ?? []), [messages]); + // Subscribe to the *derived* user history (not the raw messages array) via a + // shallow comparison: agent streaming tokens churn the messages array every + // frame but don't change the user's sent-message history, so the rich editor + // only re-renders when the history list itself changes. + return useAppStore( + useShallow((s) => + extractUserHistory((sessionId ? s.messages.bySession[sessionId] : undefined) ?? []), + ), + ); } function useChatHistory(sessionId: string | null) { diff --git a/apps/web/components/task/task-chat-panel.tsx b/apps/web/components/task/task-chat-panel.tsx index d67df35e70..d3f9018af5 100644 --- a/apps/web/components/task/task-chat-panel.tsx +++ b/apps/web/components/task/task-chat-panel.tsx @@ -143,7 +143,6 @@ export const TaskChatPanel = memo(function TaskChatPanel({ const { resolvedSessionId, session, - task, taskId, isWorking, messagesLoading, @@ -212,7 +211,6 @@ export const TaskChatPanel = memo(function TaskChatPanel({ messagesLoading={messagesLoading} isWorking={isWorking} sessionState={session?.state} - taskState={task?.state} worktreePath={session?.worktree_path} onOpenFile={onOpenFile} /> diff --git a/apps/web/components/task/task-session-sidebar-aggregate.test.ts b/apps/web/components/task/task-session-sidebar-aggregate.test.ts index e423ff3b39..f16285b058 100644 --- a/apps/web/components/task/task-session-sidebar-aggregate.test.ts +++ b/apps/web/components/task/task-session-sidebar-aggregate.test.ts @@ -1,10 +1,26 @@ import { describe, expect, it } from "vitest"; import { aggregateSidebarTasks, + buildPendingFlags, + readPendingFlags, type SidebarStepInfo, type WorkflowSnapshotMap, } from "./task-session-sidebar-aggregate"; import type { KanbanState } from "@/lib/state/slices"; +import { sessionId as toSessionId, taskId as toTaskId, type Message } from "@/lib/types/http"; + +function makePermissionRequest(id: string, status?: string): Message { + return { + id, + session_id: toSessionId("s1"), + task_id: toTaskId("t1"), + author_type: "agent", + content: "", + type: "permission_request", + metadata: status ? { status } : undefined, + created_at: "", + }; +} type KanbanTask = KanbanState["tasks"][number]; @@ -141,3 +157,31 @@ describe("aggregateSidebarTasks", () => { expect(result.allTasks.map((t) => t.id)).toEqual(["task-B"]); }); }); + +describe("buildPendingFlags / readPendingFlags", () => { + it("flags a session with a pending permission request", () => { + const flags = buildPendingFlags({ "sess-1": [makePermissionRequest("p1")] }, ["sess-1"]); + expect(readPendingFlags(flags, "sess-1")).toEqual({ clarification: false, permission: true }); + }); + + it("does not flag a session whose permission request is already resolved", () => { + const flags = buildPendingFlags({ "sess-1": [makePermissionRequest("p1", "approved")] }, [ + "sess-1", + ]); + expect(readPendingFlags(flags, "sess-1")).toEqual({ clarification: false, permission: false }); + }); + + it("only computes flags for the requested session ids", () => { + const flags = buildPendingFlags( + { "sess-1": [makePermissionRequest("p1")], "sess-2": [makePermissionRequest("p2")] }, + ["sess-1"], + ); + expect(readPendingFlags(flags, "sess-1").permission).toBe(true); + expect(readPendingFlags(flags, "sess-2").permission).toBe(false); + }); + + it("returns all-false for a null/unknown session", () => { + expect(readPendingFlags({}, null)).toEqual({ clarification: false, permission: false }); + expect(readPendingFlags({}, "missing")).toEqual({ clarification: false, permission: false }); + }); +}); diff --git a/apps/web/components/task/task-session-sidebar-aggregate.ts b/apps/web/components/task/task-session-sidebar-aggregate.ts index 12077dea2a..4c426cecb2 100644 --- a/apps/web/components/task/task-session-sidebar-aggregate.ts +++ b/apps/web/components/task/task-session-sidebar-aggregate.ts @@ -1,4 +1,39 @@ import type { KanbanState } from "@/lib/state/slices"; +import type { Message } from "@/lib/types/http"; +import { + hasPendingClarification, + hasPendingPermissionRequest, +} from "@/lib/utils/pending-clarification"; + +/** Flat per-session pending flags keyed for shallow comparison so the sidebar + * only re-renders when a clarification/permission flag actually flips, not on + * every streaming token that churns the messages map. */ +const pendingClarKey = (sessionId: string) => `${sessionId}#clar`; +const pendingPermKey = (sessionId: string) => `${sessionId}#perm`; + +export function buildPendingFlags( + bySession: Record, + sessionIds: string[], +): Record { + const flags: Record = {}; + for (const id of sessionIds) { + const msgs = bySession[id]; + flags[pendingClarKey(id)] = hasPendingClarification(msgs); + flags[pendingPermKey(id)] = hasPendingPermissionRequest(msgs); + } + return flags; +} + +export function readPendingFlags( + pendingFlags: Record, + sessionId?: string | null, +): { clarification: boolean; permission: boolean } { + if (!sessionId) return { clarification: false, permission: false }; + return { + clarification: pendingFlags[pendingClarKey(sessionId)] ?? false, + permission: pendingFlags[pendingPermKey(sessionId)] ?? false, + }; +} export type SidebarStepInfo = { id: string; diff --git a/apps/web/components/task/task-session-sidebar.tsx b/apps/web/components/task/task-session-sidebar.tsx index 33f8000236..13ddeea1e6 100644 --- a/apps/web/components/task/task-session-sidebar.tsx +++ b/apps/web/components/task/task-session-sidebar.tsx @@ -3,13 +3,7 @@ import { useCallback, useEffect, useMemo, useState, memo } from "react"; import { usePathname, useRouter } from "next/navigation"; import { linkToTask } from "@/lib/links"; -import type { - Message, - TaskState, - TaskSession, - TaskSessionState, - Repository, -} from "@/lib/types/http"; +import type { TaskState, TaskSession, TaskSessionState, Repository } from "@/lib/types/http"; import type { TaskPR } from "@/lib/types/github"; import type { KanbanState } from "@/lib/state/slices"; import type { GitStatusEntry } from "@/lib/state/slices/session-runtime/types"; @@ -33,10 +27,8 @@ import { getWebSocketClient } from "@/lib/ws/connection"; import { useArchivedTaskState } from "./task-archived-context"; import { useRepositories } from "@/hooks/domains/workspace/use-repositories"; import { useWorkspacePRs } from "@/hooks/domains/github/use-task-pr"; -import { - hasPendingClarificationForSession, - hasPendingPermissionForSession, -} from "@/lib/utils/pending-clarification"; +import { buildPendingFlags, readPendingFlags } from "./task-session-sidebar-aggregate"; +import { useShallow } from "zustand/react/shallow"; /** * Stabilize a derived array of primary session IDs so the reference only @@ -98,7 +90,7 @@ type SidebarCtx = { envIdBySessionId: Record; repositorySlugById: Map; taskPRsByTaskId: Record; - messagesBySession: Record; + pendingFlags: Record; titleById: Map; workflowNameById: Map; stepTitleById: Map; @@ -128,14 +120,7 @@ function toSidebarItem( const repoSlug = task.repositoryId ? ctx.repositorySlugById.get(task.repositoryId) : undefined; // Sidebar shows just one slot; pick the primary PR (first by created_at). const pr = ctx.taskPRsByTaskId[task.id]?.[0]; - const hasPendingClarificationRequest = hasPendingClarificationForSession( - ctx.messagesBySession, - task.primarySessionId, - ); - const hasPendingPermission = hasPendingPermissionForSession( - ctx.messagesBySession, - task.primarySessionId, - ); + const pending = readPendingFlags(ctx.pendingFlags, task.primarySessionId); const diffStats = resolveDiffStats( sessionInfo.diffStats, @@ -162,8 +147,8 @@ function toSidebarItem( remoteExecutorType: task.primaryExecutorType ?? undefined, remoteExecutorName: task.primaryExecutorName ?? undefined, primarySessionId: task.primarySessionId ?? null, - hasPendingClarification: hasPendingClarificationRequest, - hasPendingPermission, + hasPendingClarification: pending.clarification, + hasPendingPermission: pending.permission, updatedAt: sessionInfo.updatedAt ?? task.updatedAt ?? task.createdAt, createdAt: task.createdAt, isArchived: false as boolean, @@ -225,7 +210,6 @@ function useSidebarData(workspaceId: string | null) { const envIdBySessionId = useAppStore((state) => state.environmentIdBySessionId); const repositoriesByWorkspace = useAppStore((state) => state.repositories.itemsByWorkspaceId); const taskPRsByTaskId = useAppStore((state) => state.taskPRs.byTaskId); - const messagesBySession = useAppStore((state) => state.messages.bySession); const archivedState = useArchivedTaskState(); const selectedTaskId = useMemo(() => { @@ -241,6 +225,14 @@ function useSidebarData(workspaceId: string | null) { isLoading: isLoadingWorkflow, } = useWorkspaceSidebarTasks(workspaceId); + // Stable list of primary session IDs for the bulk-subscribe effect and the + // narrow pending-flag selector below. Derived from kanban tasks (always + // available) rather than sessionsByTaskId (loaded on-demand). + const primarySessionIds = useStablePrimarySessionIds(allTasks); + const pendingFlags = useAppStore( + useShallow((state) => buildPendingFlags(state.messages.bySession, primarySessionIds)), + ); + const tasksWithRepositories = useMemo(() => { const repositories = workspaceId ? (repositoriesByWorkspace[workspaceId] ?? []) : []; const repositorySlugById = new Map( @@ -255,7 +247,7 @@ function useSidebarData(workspaceId: string | null) { envIdBySessionId, repositorySlugById, taskPRsByTaskId, - messagesBySession, + pendingFlags, titleById, workflowNameById, stepTitleById, @@ -279,14 +271,10 @@ function useSidebarData(workspaceId: string | null) { gitStatusByEnvId, envIdBySessionId, taskPRsByTaskId, - messagesBySession, + pendingFlags, archivedState, ]); - // Stable list of primary session IDs for the bulk-subscribe effect. - // Derived from kanban tasks (always available) rather than sessionsByTaskId (loaded on-demand). - const primarySessionIds = useStablePrimarySessionIds(allTasks); - return { activeTaskId, selectedTaskId, diff --git a/apps/web/hooks/domains/session/use-session-messages.test.ts b/apps/web/hooks/domains/session/use-session-messages.test.ts index 78e03e1a66..870d719124 100644 --- a/apps/web/hooks/domains/session/use-session-messages.test.ts +++ b/apps/web/hooks/domains/session/use-session-messages.test.ts @@ -33,6 +33,8 @@ import { shouldRunMessageBackfill, runBackfillRound, autoBackfillUntilUserMessage, + nextFetchSeq, + commitFetchSeq, MAX_AUTO_BACKFILL_PAGES, } from "./use-session-messages"; import type { TaskSessionState } from "@/lib/types/http"; @@ -269,6 +271,32 @@ describe("runBackfillRound", () => { }); }); +describe("stale concurrent fetch guard", () => { + it("rejects an older fetch that completes after a newer one merged", () => { + const sid = "guard-sess-a"; + const older = nextFetchSeq(); + const newer = nextFetchSeq(); + // The newer fetch finishes first and merges. + expect(commitFetchSeq(sid, newer)).toBe(true); + // The older fetch finishing late must be skipped. + expect(commitFetchSeq(sid, older)).toBe(false); + }); + + it("applies fetches that complete in order", () => { + const sid = "guard-sess-b"; + expect(commitFetchSeq(sid, nextFetchSeq())).toBe(true); + expect(commitFetchSeq(sid, nextFetchSeq())).toBe(true); + }); + + it("tracks the applied sequence independently per session", () => { + const older = nextFetchSeq(); + const newer = nextFetchSeq(); + expect(commitFetchSeq("guard-sess-c", newer)).toBe(true); + // A different session is unaffected by another session's higher applied seq. + expect(commitFetchSeq("guard-sess-d", older)).toBe(true); + }); +}); + describe("autoBackfillUntilUserMessage", () => { it("continues past three pages until a user/agent message is found", async () => { mockListTaskSessionMessages diff --git a/apps/web/hooks/domains/session/use-session-messages.ts b/apps/web/hooks/domains/session/use-session-messages.ts index 239c9dfd66..a9000d1757 100644 --- a/apps/web/hooks/domains/session/use-session-messages.ts +++ b/apps/web/hooks/domains/session/use-session-messages.ts @@ -8,9 +8,35 @@ import { createDebugLogger, isDebug } from "@/lib/debug/log"; const INITIAL_FETCH_LIMIT = 100; const BACKFILL_PAGE_LIMIT = 100; const RUNNING_BACKFILL_INITIAL_DELAY_MS = 1200; -const RUNNING_BACKFILL_INTERVAL_MS = 3000; +const RUNNING_BACKFILL_INTERVAL_MS = 5000; export const MAX_AUTO_BACKFILL_PAGES = 10; +// Monotonic guard against stale concurrent fetches. Each fetch claims an +// increasing sequence number before awaiting; a completion only merges if no +// newer fetch for the same session has already merged, so an older in-flight +// fetch cannot overwrite newer state (WS-delivered messages or a newer fetch's +// snapshot) when two fetches for the same session race. +let fetchSeqCounter = 0; +const lastAppliedFetchSeq = new Map(); + +/** Claims the next monotonic fetch sequence number. */ +export function nextFetchSeq(): number { + fetchSeqCounter += 1; + return fetchSeqCounter; +} + +/** + * Records that the fetch identified by `seq` is about to merge for `sessionId`. + * Returns false when a newer fetch has already merged for the session, so the + * caller can skip the stale merge. + */ +export function commitFetchSeq(sessionId: string, seq: number): boolean { + const applied = lastAppliedFetchSeq.get(sessionId) ?? 0; + if (seq < applied) return false; + lastAppliedFetchSeq.set(sessionId, seq); + return true; +} + export function hasUserOrAgentMessage(messages: Message[]): boolean { return messages.some( (m) => m.type === "message" && (m.author_type === "user" || m.author_type === "agent"), @@ -108,6 +134,32 @@ type MessageListResponse = { messages: Message[]; has_more?: boolean; cursor?: s const EMPTY_MESSAGES: Message[] = []; const EMPTY_META = { isLoading: false, hasMore: false, oldestCursor: null }; +/** Debug-only summary of a fetch response (no-op unless debug logging is on). */ +function logFetchSummary( + sessionId: string, + fetched: Message[], + response: MessageListResponse, + limit: number, +): void { + if (!isDebug()) return; + const summary = summarizeMessages(fetched); + debug("message.list response", { + sessionId, + hasMore: response.has_more ?? false, + cursor: response.cursor ?? null, + ...summary, + }); + if (fetched.length > 0 && summary.userMessageCount === 0 && summary.agentMessageCount === 0) { + debug("WARNING: fetched window contains no user/agent message rows", { + sessionId, + limit, + hasMore: response.has_more ?? false, + byType: summary.byType, + hint: "The fetch limit may be too small for this session's last turn — user prompt and agent replies live further back. Paginate or raise the limit to see them.", + }); + } +} + /** Fetch latest messages via WS and merge with any that arrived via live notifications. */ async function fetchAndStoreMessages( sessionId: string, @@ -117,6 +169,7 @@ async function fetchAndStoreMessages( if (!client) { return []; } + const seq = nextFetchSeq(); const requestParams = { session_id: sessionId, @@ -126,24 +179,7 @@ async function fetchAndStoreMessages( debug("message.list request", requestParams); const response = await client.request("message.list", requestParams, 10000); const fetched = [...(response.messages ?? [])].reverse(); - if (isDebug()) { - const summary = summarizeMessages(fetched); - debug("message.list response", { - sessionId, - hasMore: response.has_more ?? false, - cursor: response.cursor ?? null, - ...summary, - }); - if (fetched.length > 0 && summary.userMessageCount === 0 && summary.agentMessageCount === 0) { - debug("WARNING: fetched window contains no user/agent message rows", { - sessionId, - limit: requestParams.limit, - hasMore: response.has_more ?? false, - byType: summary.byType, - hint: "The fetch limit may be too small for this session's last turn — user prompt and agent replies live further back. Paginate or raise the limit to see them.", - }); - } - } + logFetchSummary(sessionId, fetched, response, requestParams.limit); // Merge: keep WS-delivered messages that aren't in the fetch response. // This prevents a slow fetch (sent before messages existed) from wiping // messages that arrived via real-time notifications while the fetch was @@ -158,10 +194,23 @@ async function fetchAndStoreMessages( ) : fetched; - store.getState().setMessages(sessionId, merged, { + // Stale-fetch guard: if a newer fetch for this session already merged while + // this one was in flight, skip the merge so the older snapshot can't drop + // newer messages. The newer state is already in the store. + if (!commitFetchSeq(sessionId, seq)) { + debug("message.list stale fetch skipped", { sessionId, seq }); + return store.getState().messages.bySession[sessionId] ?? merged; + } + + // Idempotent merge: the store reconciles this snapshot against current state, + // preserving object/array identity for unchanged messages so the periodic + // refetch doesn't re-render the whole chat (see reconcileMessages). + store.getState().mergeMessages(sessionId, merged, { hasMore: response.has_more ?? false, oldestCursor: merged[0]?.id ?? null, }); + // The store now holds the identity-reconciled array; callers only read length + // and message content from the return, so `merged` is equivalent. return merged; } diff --git a/apps/web/hooks/use-message-navigation.test.ts b/apps/web/hooks/use-message-navigation.test.ts new file mode 100644 index 0000000000..6eb66037ae --- /dev/null +++ b/apps/web/hooks/use-message-navigation.test.ts @@ -0,0 +1,122 @@ +import { afterEach, describe, expect, it } from "vitest"; +import { createElement, type ReactNode } from "react"; +import { act, cleanup, renderHook } from "@testing-library/react"; +import { StateProvider, useAppStore } from "@/components/state-provider"; +import { sessionId as toSessionId, taskId as toTaskId, type Message } from "@/lib/types/http"; +import { useUserMessageNavigation } from "./use-message-navigation"; + +afterEach(() => cleanup()); + +const SESSION_ID = "sess-1"; + +function makeMessage(id: string, authorType: Message["author_type"]): Message { + return { + id, + session_id: toSessionId(SESSION_ID), + task_id: toTaskId("task-1"), + author_type: authorType, + content: "", + type: "message", + created_at: "2026-06-12T00:00:00Z", + }; +} + +// Render the hook alongside setMessages so a test can seed the store and read +// the resolved navigation within the same render tree. +function renderNav(currentMessageId: string, sid: string | null = SESSION_ID) { + function wrapper({ children }: { children: ReactNode }) { + return createElement(StateProvider, null, children); + } + return renderHook( + () => { + const setMessages = useAppStore((s) => s.setMessages); + const nav = useUserMessageNavigation(sid, currentMessageId); + return { setMessages, nav }; + }, + { wrapper }, + ); +} + +const NONE = { hasPrevious: false, hasNext: false, previousId: null, nextId: null }; + +type Nav = ReturnType; + +function seed(result: Nav["result"], messages: Message[]) { + act(() => result.current.setMessages(SESSION_ID, messages)); +} + +describe("useUserMessageNavigation", () => { + it("returns no navigation when the session id is null", () => { + const { result } = renderNav("u1", null); + expect(result.current.nav).toEqual(NONE); + }); + + it("returns no navigation when the session has no messages", () => { + const { result } = renderNav("u1"); + expect(result.current.nav).toEqual(NONE); + }); + + it("returns no navigation for a lone user message", () => { + const { result } = renderNav("u1"); + seed(result, [makeMessage("u1", "user")]); + expect(result.current.nav).toEqual(NONE); + }); + + it("resolves both neighbours for a middle user message", () => { + const { result } = renderNav("u2"); + seed( + result, + ["u1", "u2", "u3"].map((id) => makeMessage(id, "user")), + ); + expect(result.current.nav).toEqual({ + hasPrevious: true, + hasNext: true, + previousId: "u1", + nextId: "u3", + }); + }); + + it("resolves only next for the first user message", () => { + const { result } = renderNav("u1"); + seed( + result, + ["u1", "u2"].map((id) => makeMessage(id, "user")), + ); + expect(result.current.nav).toMatchObject({ + hasPrevious: false, + previousId: null, + nextId: "u2", + }); + }); + + it("resolves only previous for the last user message", () => { + const { result } = renderNav("u2"); + seed( + result, + ["u1", "u2"].map((id) => makeMessage(id, "user")), + ); + expect(result.current.nav).toMatchObject({ hasNext: false, previousId: "u1", nextId: null }); + }); + + it("ignores agent messages when computing neighbours", () => { + const { result } = renderNav("u2"); + seed(result, [ + makeMessage("u1", "user"), + makeMessage("a1", "agent"), + makeMessage("u2", "user"), + makeMessage("a2", "agent"), + makeMessage("u3", "user"), + ]); + expect(result.current.nav).toMatchObject({ previousId: "u1", nextId: "u3" }); + }); + + it("returns no navigation when the current message is not a user message", () => { + const { result } = renderNav("a1"); + seed(result, [ + makeMessage("u1", "user"), + makeMessage("a1", "agent"), + makeMessage("u2", "user"), + ]); + expect(result.current.nav).toEqual(NONE); + }); +}); diff --git a/apps/web/hooks/use-message-navigation.ts b/apps/web/hooks/use-message-navigation.ts index 6fd38e0cbd..ea93c202b0 100644 --- a/apps/web/hooks/use-message-navigation.ts +++ b/apps/web/hooks/use-message-navigation.ts @@ -1,35 +1,39 @@ import { useMemo } from "react"; -import type { Message } from "@/lib/types/http"; +import { useAppStore } from "@/components/state-provider"; -export function useMessageNavigation( - messages: Message[], - currentMessageId: string, - filterType: "user" | "agent", -) { - const currentIndex = useMemo(() => { - return messages.findIndex((msg) => msg.id === currentMessageId); - }, [messages, currentMessageId]); +const EMPTY_KEY = ""; - const filteredIndices = useMemo(() => { - return messages - .map((msg, idx) => (msg.author_type === filterType ? idx : -1)) - .filter((idx) => idx !== -1); - }, [messages, filterType]); +/** + * Prev/next navigation between user messages in a session, sourced directly from + * the store rather than a drilled `allMessages` array. + * + * The selector returns a value-stable key (the comma-joined ids of the session's + * user messages). During agent token streaming the user-message set is unchanged, + * so the key is referentially equal and the subscribing message does not + * re-render — which keeps `ChatMessage` memoized instead of re-running its + * markdown render on every streamed token. + */ +export function useUserMessageNavigation(sessionId: string | null, currentMessageId: string) { + const userIdsKey = useAppStore((state) => { + const messages = sessionId ? state.messages.bySession[sessionId] : undefined; + if (!messages) return EMPTY_KEY; + let key = EMPTY_KEY; + for (const message of messages) { + if (message.author_type === "user") key += `${message.id},`; + } + return key; + }); - const previous = useMemo(() => { - const validPrev = filteredIndices.filter((idx) => idx < currentIndex); - return validPrev.length > 0 ? messages[validPrev[validPrev.length - 1]] : null; - }, [filteredIndices, currentIndex, messages]); - - const next = useMemo(() => { - const validNext = filteredIndices.filter((idx) => idx > currentIndex); - return validNext.length > 0 ? messages[validNext[0]] : null; - }, [filteredIndices, currentIndex, messages]); - - return { - hasPrevious: previous !== null, - hasNext: next !== null, - previous, - next, - }; + return useMemo(() => { + const ids = userIdsKey ? userIdsKey.slice(0, -1).split(",") : []; + const index = ids.indexOf(currentMessageId); + const previousId = index > 0 ? ids[index - 1] : null; + const nextId = index >= 0 && index < ids.length - 1 ? ids[index + 1] : null; + return { + hasPrevious: previousId !== null, + hasNext: nextId !== null, + previousId, + nextId, + }; + }, [userIdsKey, currentMessageId]); } diff --git a/apps/web/hooks/use-processed-messages.test.ts b/apps/web/hooks/use-processed-messages.test.ts index 05ae204f48..b69b45ddd1 100644 --- a/apps/web/hooks/use-processed-messages.test.ts +++ b/apps/web/hooks/use-processed-messages.test.ts @@ -12,6 +12,10 @@ import { deduplicateAgentBootResumes, isAgentBootResumeMessage, isSetupScriptMessage, + messageListMapsEqual, + messageMapsEqualByIdentity, + reconcileRenderItems, + type RenderItem, } from "./use-processed-messages"; function makeMessage( @@ -257,3 +261,111 @@ describe("buildGroupedRenderItems prepare progress placement", () => { expect(result.map((item) => item.type)).toEqual(["message", "prepare_progress"]); }); }); + +describe("messageMapsEqualByIdentity", () => { + it("treats maps with the same keys and identical Message refs as equal", () => { + const a = makeMessage("perm-1", "permission_request"); + const b = makeMessage("perm-2", "permission_request"); + const first = new Map([ + ["tc-1", a], + ["tc-2", b], + ]); + const second = new Map([ + ["tc-1", a], + ["tc-2", b], + ]); + expect(messageMapsEqualByIdentity(first, second)).toBe(true); + }); + + it("is false when sizes differ", () => { + const a = makeMessage("perm-1", "permission_request"); + const first = new Map([["tc-1", a]]); + const second = new Map([ + ["tc-1", a], + ["tc-2", makeMessage("perm-2", "permission_request")], + ]); + expect(messageMapsEqualByIdentity(first, second)).toBe(false); + }); + + it("is false when a key maps to a different Message ref (e.g. status changed)", () => { + const first = new Map([["tc-1", makeMessage("perm-1", "permission_request")]]); + const second = new Map([ + ["tc-1", makeMessage("perm-1", "permission_request")], + ]); + expect(messageMapsEqualByIdentity(first, second)).toBe(false); + }); +}); + +describe("messageListMapsEqual", () => { + it("treats maps whose array values share positional Message refs as equal", () => { + const c1 = makeMessage("child-1", "tool_call"); + const c2 = makeMessage("child-2", "tool_call"); + const first = new Map([["parent-1", [c1, c2]]]); + const second = new Map([["parent-1", [c1, c2]]]); + expect(messageListMapsEqual(first, second)).toBe(true); + }); + + it("is false when an array length changes (new child appended)", () => { + const c1 = makeMessage("child-1", "tool_call"); + const first = new Map([["parent-1", [c1]]]); + const second = new Map([ + ["parent-1", [c1, makeMessage("child-2", "tool_call")]], + ]); + expect(messageListMapsEqual(first, second)).toBe(false); + }); + + it("is false when a child Message ref changes positionally", () => { + const c1 = makeMessage("child-1", "tool_call"); + const first = new Map([["parent-1", [c1]]]); + const second = new Map([ + ["parent-1", [makeMessage("child-1", "tool_call")]], + ]); + expect(messageListMapsEqual(first, second)).toBe(false); + }); +}); + +describe("reconcileRenderItems", () => { + const turnGroup = (id: string, messages: Message[]): RenderItem => ({ + type: "turn_group", + id, + turnId: messages[0]?.turn_id ?? null, + messages, + }); + const messageItem = (message: Message): RenderItem => ({ type: "message", message }); + + it("returns the prior array unchanged when nothing changed", () => { + const m1 = makeMessage("m1", "message"); + const prev = [messageItem(m1)]; + const next = [messageItem(m1)]; + expect(reconcileRenderItems(prev, next)).toBe(prev); + }); + + it("reuses unchanged turn-group wrappers while replacing the changed one", () => { + const a1 = makeMessage("a1", "tool_call"); + const a2 = makeMessage("a2", "tool_call"); + const b1 = makeMessage("b1", "tool_call"); + const b2 = makeMessage("b2", "tool_call"); + const groupA = turnGroup("turn-group-a1", [a1, a2]); + const groupBPrev = turnGroup("turn-group-b1", [b1]); + const prev = [groupA, groupBPrev]; + + // groupA identical content (rebuilt wrapper); groupB grows by a token. + const groupBNext = turnGroup("turn-group-b1", [b1, b2]); + const next = [turnGroup("turn-group-a1", [a1, a2]), groupBNext]; + + const result = reconcileRenderItems(prev, next); + expect(result[0]).toBe(groupA); // stable → memo holds + expect(result[1]).toBe(groupBNext); // changed → fresh wrapper + }); + + it("keeps stable message wrappers and appends a new item", () => { + const m1 = makeMessage("m1", "message"); + const prev = [messageItem(m1)]; + const m2 = makeMessage("m2", "message"); + const next = [messageItem(m1), messageItem(m2)]; + const result = reconcileRenderItems(prev, next); + expect(result).not.toBe(prev); + expect(result[0]).toBe(prev[0]); + expect(result).toHaveLength(2); + }); +}); diff --git a/apps/web/hooks/use-processed-messages.ts b/apps/web/hooks/use-processed-messages.ts index 537161c2b7..e34e31ffa3 100644 --- a/apps/web/hooks/use-processed-messages.ts +++ b/apps/web/hooks/use-processed-messages.ts @@ -1,4 +1,4 @@ -import { useEffect, useMemo } from "react"; +import { useEffect, useMemo, useState } from "react"; import { sessionId as toSessionId, taskId as toTaskId, @@ -157,6 +157,98 @@ function buildSubagentChildIds(childrenByParentToolCallId: Map, + b: Map, +): boolean { + if (a.size !== b.size) return false; + for (const [key, value] of a) { + if (b.get(key) !== value) return false; + } + return true; +} + +/** True when both maps hold the same keys pointing at arrays whose Message refs + * match positionally. */ +export function messageListMapsEqual( + a: Map, + b: Map, +): boolean { + if (a.size !== b.size) return false; + for (const [key, value] of a) { + const other = b.get(key); + if (!other || other.length !== value.length) return false; + for (let i = 0; i < value.length; i++) { + if (value[i] !== other[i]) return false; + } + } + return true; +} + +/** Keep a prior reference when the freshly-built value is content-equal. The + * per-token `messages` array churns every reference downstream; holding the + * prior value lets `memo` on each message component survive commits where the + * derived map content didn't actually change. Uses the React-endorsed + * "adjust state during render" pattern so we never mutate a ref mid-render. */ +function useStableValue(next: T, equal: (a: T, b: T) => boolean): T { + const [stable, setStable] = useState(next); + if (!equal(stable, next)) { + setStable(next); + return next; + } + return stable; +} + +function renderItemKey(item: RenderItem): string { + return item.type === "message" ? `m:${item.message.id}` : item.id; +} + +/** Content-equal when the wrapper would render identically. Turn groups compare + * their messages by positional identity so a streaming token in the active turn + * yields a fresh wrapper while quiescent earlier turns stay equal. */ +function renderItemsEqual(a: RenderItem, b: RenderItem): boolean { + if (a.type !== b.type) return false; + if (a.type === "message" && b.type === "message") return a.message === b.message; + if (a.type === "prepare_progress" && b.type === "prepare_progress") return a.id === b.id; + if (a.type === "turn_group" && b.type === "turn_group") { + if (a.id !== b.id || a.turnId !== b.turnId || a.messages.length !== b.messages.length) { + return false; + } + for (let i = 0; i < a.messages.length; i++) { + if (a.messages[i] !== b.messages[i]) return false; + } + return true; + } + return false; +} + +/** Reuse prior wrapper objects for items whose content didn't change so the + * `memo` on TurnGroupMessage/MessageRenderer survives new-message commits. + * Returns `prev` unchanged when every slot is positionally identical. */ +export function reconcileRenderItems(prev: RenderItem[], next: RenderItem[]): RenderItem[] { + const prevByKey = new Map(); + for (const item of prev) prevByKey.set(renderItemKey(item), item); + let identical = prev.length === next.length; + const result = next.map((nextItem, i) => { + const prevItem = prevByKey.get(renderItemKey(nextItem)); + const reused = prevItem && renderItemsEqual(prevItem, nextItem) ? prevItem : nextItem; + if (reused !== prev[i]) identical = false; + return reused; + }); + return identical ? prev : result; +} + +function useStableGroupedItems(next: RenderItem[]): RenderItem[] { + const [stable, setStable] = useState(next); + const reconciled = reconcileRenderItems(stable, next); + if (reconciled !== stable) { + setStable(reconciled); + return reconciled; + } + return stable; +} + function isRecoveryMessage(message: Message): boolean { const meta = message.metadata as Record | undefined; return meta?.recovery_actions === true; @@ -407,10 +499,13 @@ export function useProcessedMessages( options: ProcessedMessagesOptions = {}, ) { const toolCallIds = useMemo(() => buildToolCallIds(messages), [messages]); - const permissionsByToolCallId = useMemo(() => buildPermissionsByToolCallId(messages), [messages]); - const childrenByParentToolCallId = useMemo( - () => buildChildrenByParentToolCallId(messages), - [messages], + const permissionsByToolCallId = useStableValue( + useMemo(() => buildPermissionsByToolCallId(messages), [messages]), + messageMapsEqualByIdentity, + ); + const childrenByParentToolCallId = useStableValue( + useMemo(() => buildChildrenByParentToolCallId(messages), [messages]), + messageListMapsEqual, ); const subagentChildIds = useMemo( () => buildSubagentChildIds(childrenByParentToolCallId), @@ -467,7 +562,7 @@ export function useProcessedMessages( return { regularMessages: regular, footerActionMessages: footer }; }, [allMessages]); - const groupedItems = useMemo(() => { + const rawGroupedItems = useMemo(() => { return buildGroupedItemsForHook({ regularMessages, allSessionMessages: messages, @@ -475,6 +570,7 @@ export function useProcessedMessages( hasOlderMessages: options.hasOlderMessages ?? false, }); }, [regularMessages, resolvedSessionId, messages, options.hasOlderMessages]); + const groupedItems = useStableGroupedItems(rawGroupedItems); useDebugProcessedPipeline({ sessionId: resolvedSessionId, diff --git a/apps/web/lib/markdown/normalize-cache.test.ts b/apps/web/lib/markdown/normalize-cache.test.ts new file mode 100644 index 0000000000..50e18d7093 --- /dev/null +++ b/apps/web/lib/markdown/normalize-cache.test.ts @@ -0,0 +1,67 @@ +import { beforeEach, describe, expect, it } from "vitest"; +import { + __MAX_CACHE_ENTRIES, + __lruSize, + __markdownParseCount, + __resetMarkdownCounters, + normalizeCached, + normalizeMarkdown, +} from "./normalize-cache"; + +describe("normalizeCached", () => { + beforeEach(() => { + __resetMarkdownCounters(); + }); + + it("parses once for repeated identical input (cache hit)", () => { + const input = "```go\nfunc f() {}```\nprose"; + const first = normalizeCached(input); + const second = normalizeCached(input); + expect(second).toBe(first); + expect(__markdownParseCount()).toBe(1); + }); + + it("parses again for different input", () => { + normalizeCached("alpha"); + normalizeCached("beta"); + expect(__markdownParseCount()).toBe(2); + }); + + it("produces output byte-identical to normalizeMarkdown", () => { + const inputs = [ + "```go\nfunc f() {}```\nprose", + "Use `code` inline.", + "", + "single line", + "```go\nx```\nprose\n", + ]; + for (const input of inputs) { + expect(normalizeCached(input)).toBe(normalizeMarkdown(input)); + } + }); + + it("evicts oldest entries past the cap (bounded LRU)", () => { + for (let i = 0; i < __MAX_CACHE_ENTRIES + 50; i++) { + normalizeCached(`unique-content-${i}`); + } + expect(__lruSize()).toBeLessThanOrEqual(__MAX_CACHE_ENTRIES); + }); + + it("keeps a recently-used entry warm despite overflow", () => { + normalizeCached("keep-me"); + for (let i = 0; i < __MAX_CACHE_ENTRIES; i++) { + normalizeCached(`filler-${i}`); + normalizeCached("keep-me"); // refresh recency each round + } + const before = __markdownParseCount(); + normalizeCached("keep-me"); + expect(__markdownParseCount()).toBe(before); // still cached, no new parse + }); + + it("__resetMarkdownCounters clears cache and counter", () => { + normalizeCached("something"); + __resetMarkdownCounters(); + expect(__markdownParseCount()).toBe(0); + expect(__lruSize()).toBe(0); + }); +}); diff --git a/apps/web/lib/markdown/normalize-cache.ts b/apps/web/lib/markdown/normalize-cache.ts new file mode 100644 index 0000000000..91c948a2d1 --- /dev/null +++ b/apps/web/lib/markdown/normalize-cache.ts @@ -0,0 +1,125 @@ +/** + * Pure markdown normalization plus a bounded, value-keyed LRU cache. + * + * `normalizeMarkdown` is a pure string transform (moved here from + * markdown-components so the cache has no React dependency). `normalizeCached` + * wraps it in a hard-capped LRU keyed by the raw input string, so repeated + * renders of the same content never re-run the transform. The cache is keyed by + * content only (never by task/session/message id), so it cannot leak per-task + * state: entries are anonymous strings that age out via LRU. + */ + +const FENCE_OPEN_RE = /^ {0,3}(`{3,})/; +const TRAILING_FENCE_RE = /(`{3,})\s*$/; + +function pureCloseLength(line: string, openCount: number): number | null { + const match = /^ {0,3}(`{3,})\s*$/.exec(line); + if (!match || match[1].length < openCount) return null; + return match[1].length; +} + +function gluedCloseLength(line: string, openCount: number): number | null { + const match = TRAILING_FENCE_RE.exec(line); + if (!match || match[1].length < openCount) return null; + // Reject pure-fence lines (already handled by pureCloseLength) and lines + // where everything before the trailing run is whitespace only. + const head = line.slice(0, line.length - match[0].length).trimEnd(); + if (head.length === 0) return null; + return match[1].length; +} + +/** + * Pre-process a markdown string to repair fenced code blocks that have their + * closing fence glued to the last code line (`...}\`\`\`\n`prose`). Without + * this, CommonMark/GFM treats the glued backticks as code content, so the + * fence never closes and following prose gets swallowed into one huge code + * node. We split such lines into `\n` only when we're + * inside an open fence whose opener run length is ≤ the trailing run length. + * + * Pure string preprocessing, intentionally not a remark plugin. + */ +export function normalizeMarkdown(input: string): string { + if (!input || input.length === 0) return input; + const hadTrailingNewline = input.endsWith("\n"); + const lines = input.split("\n"); + const out: string[] = []; + let openCount: number | null = null; + + for (const line of lines) { + if (openCount === null) { + const opener = FENCE_OPEN_RE.exec(line); + if (opener) openCount = opener[1].length; + out.push(line); + continue; + } + if (pureCloseLength(line, openCount) !== null) { + openCount = null; + out.push(line); + continue; + } + const glued = gluedCloseLength(line, openCount); + if (glued !== null) { + const trailingMatch = TRAILING_FENCE_RE.exec(line)!; + const head = line.slice(0, line.length - trailingMatch[0].length); + out.push(head); + out.push("`".repeat(glued)); + openCount = null; + continue; + } + out.push(line); + } + + const result = out.join("\n"); + return hadTrailingNewline && !result.endsWith("\n") ? result + "\n" : result; +} + +// ── Bounded LRU cache ─────────────────────────────────────────────── + +const MAX_CACHE_ENTRIES = 500; +const normalizeCache = new Map(); +let parseCount = 0; + +/** + * Cached variant of {@link normalizeMarkdown}. Keyed by the raw input string; + * a hit refreshes recency (Map preserves insertion order, so delete+set moves + * the entry to the end). On overflow the oldest entry is evicted. + */ +export function normalizeCached(input: string): string { + const cached = normalizeCache.get(input); + if (cached !== undefined) { + normalizeCache.delete(input); + normalizeCache.set(input, cached); + return cached; + } + parseCount += 1; + const result = normalizeMarkdown(input); + // Evict before insert so the map never exceeds the cap. A miss means the key + // isn't present, so the cache holds at most MAX_CACHE_ENTRIES - 1 entries here. + if (normalizeCache.size >= MAX_CACHE_ENTRIES) { + const oldest = normalizeCache.keys().next().value; + if (oldest !== undefined) normalizeCache.delete(oldest); + } + normalizeCache.set(input, result); + return result; +} + +// ── Test-only instrumentation ─────────────────────────────────────── + +/** Number of real normalize passes (cache misses) since the last reset. */ +export function __markdownParseCount(): number { + return parseCount; +} + +/** Clears the cache and resets the parse counter. Test-only. */ +export function __resetMarkdownCounters(): void { + parseCount = 0; + normalizeCache.clear(); +} + +/** Current cache size. Test-only (asserts the LRU stays bounded). */ +export function __lruSize(): number { + return normalizeCache.size; +} + +/** Hard cap on cached entries. Test-only. */ +export const __MAX_CACHE_ENTRIES = MAX_CACHE_ENTRIES; diff --git a/apps/web/lib/state/slices/session/message-signature.test.ts b/apps/web/lib/state/slices/session/message-signature.test.ts new file mode 100644 index 0000000000..6d404bcffc --- /dev/null +++ b/apps/web/lib/state/slices/session/message-signature.test.ts @@ -0,0 +1,130 @@ +import { describe, expect, it } from "vitest"; +import { sessionId as toSessionId, taskId as toTaskId, type Message } from "@/lib/types/http"; +import { reconcileMessages, signatureOf } from "./message-signature"; + +const TS = "2024-01-01T00:00:00Z"; + +function makeMessage(overrides: Partial): Message { + return { + id: "msg-1", + task_id: toTaskId("task-1"), + session_id: toSessionId("sess-1"), + author_type: "user", + content: "hello", + type: "message", + created_at: TS, + ...overrides, + } as Message; +} + +describe("signatureOf", () => { + it("returns equal signatures for messages with equal content fields", () => { + const a = makeMessage({ content: "one", metadata: { foo: { bar: 1 } } }); + const b = makeMessage({ content: "one", metadata: { foo: { bar: 1 } } }); + expect(signatureOf(a)).toBe(signatureOf(b)); + }); + + it("differs when content changes", () => { + expect(signatureOf(makeMessage({ content: "one" }))).not.toBe( + signatureOf(makeMessage({ content: "two" })), + ); + }); + + it("differs when metadata changes", () => { + expect(signatureOf(makeMessage({ metadata: { a: 1 } }))).not.toBe( + signatureOf(makeMessage({ metadata: { a: 2 } })), + ); + }); + + it("is stable regardless of metadata key insertion order", () => { + const a = makeMessage({ metadata: { a: 1, b: 2 } }); + const b = makeMessage({ metadata: { b: 2, a: 1 } }); + expect(signatureOf(a)).toBe(signatureOf(b)); + }); + + it("differs when turn_id or requests_input change", () => { + const base = makeMessage({ content: "x" }); + expect(signatureOf(base)).not.toBe(signatureOf(makeMessage({ content: "x", turn_id: "t1" }))); + expect(signatureOf(base)).not.toBe( + signatureOf(makeMessage({ content: "x", requests_input: true })), + ); + }); + + it("caches per object (same object returns identical signature)", () => { + const m = makeMessage({ content: "cached" }); + expect(signatureOf(m)).toBe(signatureOf(m)); + }); + + it("short-circuits to updated_at when present (authoritative signal)", () => { + const a = makeMessage({ content: "x", updated_at: TS }); + const b = makeMessage({ content: "different", updated_at: TS }); + expect(signatureOf(a)).toBe(signatureOf(b)); + }); + + it("differs when updated_at advances", () => { + const a = makeMessage({ content: "x", updated_at: TS }); + const b = makeMessage({ content: "x", updated_at: "2024-01-01T00:00:01Z" }); + expect(signatureOf(a)).not.toBe(signatureOf(b)); + }); + + it("falls back to content hash when updated_at is absent", () => { + expect(signatureOf(makeMessage({ content: "one" }))).not.toBe( + signatureOf(makeMessage({ content: "two" })), + ); + }); +}); + +describe("reconcileMessages", () => { + it("returns the prev array unchanged when every message is signature-equal", () => { + const prev = [ + makeMessage({ id: "a", content: "one" }), + makeMessage({ id: "b", content: "two", metadata: { foo: { bar: 1 } } }), + ]; + const next = [ + makeMessage({ id: "a", content: "one" }), + makeMessage({ id: "b", content: "two", metadata: { foo: { bar: 1 } } }), + ]; + expect(reconcileMessages(prev, next)).toBe(prev); + }); + + it("reuses prior references for unchanged messages and keeps the changed one fresh", () => { + const prev = [ + makeMessage({ id: "a", content: "one" }), + makeMessage({ id: "b", content: "two" }), + ]; + const next = [ + makeMessage({ id: "a", content: "one" }), + makeMessage({ id: "b", content: "two-edited" }), + ]; + const result = reconcileMessages(prev, next); + expect(result).not.toBe(prev); + expect(result[0]).toBe(prev[0]); + expect(result[1]).toBe(next[1]); + }); + + it("returns next when a message is appended, reusing existing references", () => { + const prev = [makeMessage({ id: "a", content: "one" })]; + const next = [ + makeMessage({ id: "a", content: "one" }), + makeMessage({ id: "b", content: "two" }), + ]; + const result = reconcileMessages(prev, next); + expect(result).not.toBe(prev); + expect(result[0]).toBe(prev[0]); + expect(result[1]).toBe(next[1]); + }); + + it("returns next directly when prev is empty or undefined", () => { + const next = [makeMessage({ id: "a" })]; + expect(reconcileMessages([], next)).toBe(next); + expect(reconcileMessages(undefined, next)).toBe(next); + }); + + it("detects a metadata-only change", () => { + const prev = [makeMessage({ id: "a", content: "one", metadata: { x: 1 } })]; + const next = [makeMessage({ id: "a", content: "one", metadata: { x: 2 } })]; + const result = reconcileMessages(prev, next); + expect(result).not.toBe(prev); + expect(result[0]).toBe(next[0]); + }); +}); diff --git a/apps/web/lib/state/slices/session/message-signature.ts b/apps/web/lib/state/slices/session/message-signature.ts new file mode 100644 index 0000000000..eedbcbd819 --- /dev/null +++ b/apps/web/lib/state/slices/session/message-signature.ts @@ -0,0 +1,83 @@ +import type { Message } from "@/lib/types/http"; + +const SEP = "\u0000"; +const signatureCache = new WeakMap(); + +/** + * djb2 string hash → unsigned base36. Non-crypto; paired with a length prefix + * and the message id (during reconciliation) so collisions cannot cause a + * visible reuse of a changed message. + */ +function hashString(input: string): string { + let hash = 5381; + for (let i = 0; i < input.length; i++) { + hash = ((hash << 5) + hash) ^ input.charCodeAt(i); + } + return (hash >>> 0).toString(36); +} + +function serializeMetadata(metadata: Record | undefined): string { + if (!metadata) return ""; + // Sort keys so signature is stable regardless of insertion order. + const keys = Object.keys(metadata).sort(); + let out = ""; + for (const key of keys) out += `${key}=${JSON.stringify(metadata[key])}${SEP}`; + return out; +} + +function contentHashSignature(message: Message): string { + const parts = + message.content + + SEP + + message.type + + SEP + + (message.turn_id ?? "") + + SEP + + (message.requests_input ? "1" : "0") + + SEP + + (message.raw_content ?? "") + + SEP + + serializeMetadata(message.metadata); + return `h:${parts.length}:${hashString(parts)}`; +} + +/** + * Content signature of a message, cached per-object via a WeakMap so each + * Message object is hashed at most once. Two messages with the same id and the + * same signature are treated as unchanged for reconciliation. The WeakMap is + * keyed by the Message object, so entries are GC'd once messages leave the + * store (e.g. task delete/archive) — no cap or cleanup hook needed. + * + * When the backend supplies `updated_at` (the authoritative per-message change + * signal, bumped on every content/metadata mutation including streaming tokens), + * the signature short-circuits to it — an O(1) compare with no content hashing. + * Older payloads without `updated_at` fall back to the content hash. + */ +export function signatureOf(message: Message): string { + const cached = signatureCache.get(message); + if (cached !== undefined) return cached; + const signature = message.updated_at ? `u:${message.updated_at}` : contentHashSignature(message); + signatureCache.set(message, signature); + return signature; +} + +/** + * Reconcile a full snapshot against the previous array, preserving object + * identity for messages whose signature is unchanged and returning the prev + * array reference itself when nothing moved (so array identity is preserved and + * no downstream re-render fires). Only genuinely-changed messages get a fresh + * reference. + */ +export function reconcileMessages(prev: Message[] | undefined, next: Message[]): Message[] { + if (!prev || prev.length === 0) return next; + const prevById = new Map(); + for (const message of prev) prevById.set(message.id, message); + let identical = prev.length === next.length; + const result = next.map((message, i) => { + const previous = prevById.get(message.id); + const reused = previous && signatureOf(previous) === signatureOf(message) ? previous : message; + if (reused !== prev[i]) identical = false; + return reused; + }); + return identical ? prev : result; +} diff --git a/apps/web/lib/state/slices/session/session-slice.merge-messages.test.ts b/apps/web/lib/state/slices/session/session-slice.merge-messages.test.ts new file mode 100644 index 0000000000..4c91392949 --- /dev/null +++ b/apps/web/lib/state/slices/session/session-slice.merge-messages.test.ts @@ -0,0 +1,97 @@ +import { describe, it, expect } from "vitest"; +import { create } from "zustand"; +import { immer } from "zustand/middleware/immer"; +import { createSessionSlice } from "./session-slice"; +import type { SessionSlice } from "./types"; +import { sessionId as toSessionId, taskId as toTaskId, type Message } from "@/lib/types/http"; + +function makeStore() { + return create()( + // eslint-disable-next-line @typescript-eslint/no-explicit-any + immer((set) => ({ ...(createSessionSlice as any)(set) })), + ); +} + +const SESSION = "sess-1"; + +function makeMessage(overrides: Partial): Message { + return { + id: "m1", + task_id: toTaskId("task-1"), + session_id: toSessionId(SESSION), + author_type: "user", + content: "hello", + type: "message", + created_at: "2024-01-01T00:00:00Z", + ...overrides, + } as Message; +} + +function snapshot(): Message[] { + return [makeMessage({ id: "a", content: "one" }), makeMessage({ id: "b", content: "two" })]; +} + +describe("mergeMessages", () => { + it("preserves the array reference on a no-op refetch", () => { + const store = makeStore(); + store.getState().mergeMessages(SESSION, snapshot()); + const first = store.getState().messages.bySession[SESSION]; + // Fresh, equal objects as a refetch would produce. + store.getState().mergeMessages(SESSION, snapshot()); + expect(store.getState().messages.bySession[SESSION]).toBe(first); + }); + + it("changes the array ref and only the changed element when one message changes", () => { + const store = makeStore(); + store.getState().mergeMessages(SESSION, snapshot()); + const first = store.getState().messages.bySession[SESSION]; + + store + .getState() + .mergeMessages(SESSION, [ + makeMessage({ id: "a", content: "one" }), + makeMessage({ id: "b", content: "two-edited" }), + ]); + const next = store.getState().messages.bySession[SESSION]; + expect(next).not.toBe(first); + expect(next[0]).toBe(first[0]); // unchanged message kept its identity + expect(next[1]).not.toBe(first[1]); + expect(next[1].content).toBe("two-edited"); + }); + + it("appends new messages while reusing existing references", () => { + const store = makeStore(); + store.getState().mergeMessages(SESSION, [makeMessage({ id: "a", content: "one" })]); + const first = store.getState().messages.bySession[SESSION]; + + store + .getState() + .mergeMessages(SESSION, [ + makeMessage({ id: "a", content: "one" }), + makeMessage({ id: "b", content: "two" }), + ]); + const next = store.getState().messages.bySession[SESSION]; + expect(next).not.toBe(first); + expect(next[0]).toBe(first[0]); + expect(next).toHaveLength(2); + }); + + it("detects a metadata-only change", () => { + const store = makeStore(); + store.getState().mergeMessages(SESSION, [makeMessage({ id: "a", metadata: { x: 1 } })]); + const first = store.getState().messages.bySession[SESSION]; + + store.getState().mergeMessages(SESSION, [makeMessage({ id: "a", metadata: { x: 2 } })]); + const next = store.getState().messages.bySession[SESSION]; + expect(next).not.toBe(first); + expect((next[0].metadata as { x: number }).x).toBe(2); + }); + + it("applies metadata (hasMore / oldestCursor) on merge", () => { + const store = makeStore(); + store.getState().mergeMessages(SESSION, snapshot(), { hasMore: true, oldestCursor: "a" }); + const meta = store.getState().messages.metaBySession[SESSION]; + expect(meta.hasMore).toBe(true); + expect(meta.oldestCursor).toBe("a"); + }); +}); diff --git a/apps/web/lib/state/slices/session/session-slice.ts b/apps/web/lib/state/slices/session/session-slice.ts index 158b7c5db8..c7aeea4d44 100644 --- a/apps/web/lib/state/slices/session/session-slice.ts +++ b/apps/web/lib/state/slices/session/session-slice.ts @@ -1,6 +1,8 @@ import type { StateCreator } from "zustand"; -import type { TaskSession } from "@/lib/types/http"; +import { original } from "immer"; +import type { Message, TaskSession } from "@/lib/types/http"; import type { SessionSlice, SessionSliceState } from "./types"; +import { reconcileMessages } from "./message-signature"; import { migrateEnvKeyedData } from "@/lib/state/slices/session-runtime/session-runtime-slice"; import { prepareResultToSessionState } from "@/lib/state/slices/session-runtime/prepare-result"; import { createDebugLogger, isDebug } from "@/lib/debug/log"; @@ -175,6 +177,25 @@ function buildMessageActions(set: ImmerSet) { ); messages[index] = merged; }), + mergeMessages: ( + sessionId: string, + messages: Parameters[1], + meta?: Parameters[2], + ) => + set((draft) => { + const prevDraft = draft.messages.bySession[sessionId]; + const prev = (prevDraft ? (original(prevDraft) ?? prevDraft) : undefined) as + | Message[] + | undefined; + const reconciled = reconcileMessages(prev, messages); + // Only replace the array when identity actually changed, so a no-op + // refetch preserves the array reference and triggers no re-render. + if (reconciled !== prev) { + draft.messages.bySession[sessionId] = reconciled; + } + ensureMessageMeta(draft.messages.metaBySession, sessionId); + if (meta) applyMessageMeta(draft.messages.metaBySession, sessionId, meta); + }), prependMessages: ( sessionId: string, messages: Parameters[1], diff --git a/apps/web/lib/state/slices/session/types.ts b/apps/web/lib/state/slices/session/types.ts index e6398753cf..0de1094b8e 100644 --- a/apps/web/lib/state/slices/session/types.ts +++ b/apps/web/lib/state/slices/session/types.ts @@ -150,6 +150,17 @@ export type SessionSliceActions = { ) => void; addMessage: (message: Message) => void; updateMessage: (message: Message) => void; + /** + * Idempotent full-snapshot merge: reconciles `messages` against the current + * stored array, preserving object identity for unchanged messages and the + * array reference itself when nothing changed (see `reconcileMessages`). Used + * by periodic refetches so a no-op tick triggers zero re-renders. + */ + mergeMessages: ( + sessionId: string, + messages: Message[], + meta?: { hasMore?: boolean; oldestCursor?: string | null }, + ) => void; prependMessages: ( sessionId: string, messages: Message[], diff --git a/apps/web/lib/state/store.ts b/apps/web/lib/state/store.ts index 6979c7c834..5524b31ff3 100644 --- a/apps/web/lib/state/store.ts +++ b/apps/web/lib/state/store.ts @@ -387,6 +387,11 @@ export type AppState = { meta?: { hasMore?: boolean; oldestCursor?: string | null }, ) => void; addMessage: (message: Message) => void; + mergeMessages: ( + sessionId: string, + messages: Message[], + meta?: { hasMore?: boolean; oldestCursor?: string | null }, + ) => void; addTurn: (turn: Turn) => void; completeTurn: (sessionId: string, turnId: string, completedAt: string) => void; setActiveTurn: (sessionId: string, turnId: string | null) => void; diff --git a/apps/web/lib/types/backend.ts b/apps/web/lib/types/backend.ts index e3f71d2ffb..4283fbfaed 100644 --- a/apps/web/lib/types/backend.ts +++ b/apps/web/lib/types/backend.ts @@ -256,6 +256,7 @@ export type MessageAddedPayload = { metadata?: Record; requests_input?: boolean; created_at: string; + updated_at?: string; }; export type TaskSessionStateChangedPayload = { diff --git a/apps/web/lib/types/http.ts b/apps/web/lib/types/http.ts index 0bb331bd6c..07ddddc45b 100644 --- a/apps/web/lib/types/http.ts +++ b/apps/web/lib/types/http.ts @@ -663,6 +663,8 @@ export type Message = { metadata?: Record; requests_input?: boolean; created_at: string; + /** Authoritative per-message change signal; advances on every content/metadata update. */ + updated_at?: string; }; export type Turn = { diff --git a/apps/web/lib/ws/handlers/messages.ts b/apps/web/lib/ws/handlers/messages.ts index 9981c56fca..6ac7685b10 100644 --- a/apps/web/lib/ws/handlers/messages.ts +++ b/apps/web/lib/ws/handlers/messages.ts @@ -23,6 +23,7 @@ export function registerMessagesHandlers(store: StoreApi): WsHandlers metadata: payload.metadata, requests_input: payload.requests_input, created_at: payload.created_at, + updated_at: payload.updated_at, }); }, "session.message.updated": (message) => { @@ -43,6 +44,7 @@ export function registerMessagesHandlers(store: StoreApi): WsHandlers metadata: payload.metadata, requests_input: payload.requests_input, created_at: payload.created_at, + updated_at: payload.updated_at, }); }, };